diff --git a/.github/workflows/php.yml b/.github/workflows/php.yml index d798ee71f..472172e69 100644 --- a/.github/workflows/php.yml +++ b/.github/workflows/php.yml @@ -174,7 +174,9 @@ jobs: regression-corpus: needs: preflight runs-on: ubuntu-latest - timeout-minutes: 6 + # Each new fixture adds two full consumer runs after the baseline. + # Four fixtures already exceed six minutes on the hosted runner. + timeout-minutes: 12 env: # The runner's system PHP loads Xdebug. Each official consumer run takes # several times longer with it, and the validator runs the consumer diff --git a/CHANGELOG.md b/CHANGELOG.md index 769aa0026..ef222ece1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,26 @@ ## Unreleased +- **Run migrations.** Opt-in deadline-triggered cleanup (#863, experimental, + embedded). A start can reserve the last part of its hard execution or run + timeout for cleanup with `StartOptions::withTimeoutCleanup($seconds)` (or + `timeoutCleanupSeconds:`). When the work deadline passes (the hard deadline + minus the allowance), the run enters timeout cleanup once: workflow code + receives `WorkflowTimeoutCleanupRequested`, a cooperative cancellation + request whose context has `cause: timeout`, so existing compensation + handlers run. An idle run is woken at that time. The run always ends with + `WorkflowTimedOut`, which now carries a `timeout_cleanup` outcome: + `completed`, `failed` (the cleanup failure is linked), `expired` (still + running at the hard deadline) or `preempted` (terminated or hard-cancelled + first). The hard deadline never moves, and nothing changes without the + option. Continue-as-new and redrive carry the allowance like the hard + deadlines. New history event `WorkflowTimeoutCleanupStarted`; the cancellation + context gains the optional top-level `cause`. The migration adds the nullable + `workflow_runs.timeout_cleanup_work_deadline_at` with no backfill and builds + its index without blocking writes. Poll-mode and service starts refuse this + option until their capability admission is implemented (server#1201). Bridge + workers cannot claim an opted-in embedded run. Service mode follows separately. + ## 2.28.0 - 2026-11-13 **Behavior change:** an un-awaited `select()` loser is cancelled when the run diff --git a/docs/api-stability.md b/docs/api-stability.md index d879db446..e9618b03f 100644 --- a/docs/api-stability.md +++ b/docs/api-stability.md @@ -765,9 +765,10 @@ class, source file path, or stack trace to replay or handle a failure. | --- | --- | --- | | `StartAccepted` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `workflow_class`, `workflow_type`, `business_key`, `visibility_labels`, `memo`, `search_attributes`, `outcome`, `rejection_reason` | `HistoryTimeline`, `HistoryExport`, `RunCommandContract`, operator command projections | | `StartRejected` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `workflow_class`, `workflow_type`, `business_key`, `visibility_labels`, `memo`, `search_attributes`, `outcome`, `rejection_reason` | `HistoryTimeline`, `HistoryExport`, `RunCommandContract`, operator command projections | -| `WorkflowStarted` | `workflow_class`, `workflow_type`, `workflow_instance_id`, `workflow_run_id`, `workflow_command_id`, `business_key`, `visibility_labels`, `memo`, `search_attributes`, `execution_timeout_seconds`, `run_timeout_seconds`, `execution_deadline_at`, `run_deadline_at`, `workflow_definition_fingerprint`, `definition_changed_from`, `declared_queries`, `declared_query_contracts`, `declared_signals`, `declared_signal_contracts`, `declared_updates`, `declared_update_contracts`, `declared_entry_method`, `declared_entry_mode`, `declared_entry_declaring_class`, `parent_workflow_instance_id`, `parent_workflow_run_id`, `parent_sequence`, `workflow_link_id`, `child_call_id`, `retry_policy`, `timeout_policy`, `continued_from_run_id`, `retry_attempt`, `retry_of_child_workflow_run_id` | `WorkflowDefinitionFingerprint`, `RunLineageView`, worker history payload consumers | +| `WorkflowStarted` | `workflow_class`, `workflow_type`, `workflow_instance_id`, `workflow_run_id`, `workflow_command_id`, `business_key`, `visibility_labels`, `memo`, `search_attributes`, `execution_timeout_seconds`, `run_timeout_seconds`, `execution_deadline_at`, `run_deadline_at`, `workflow_definition_fingerprint`, `definition_changed_from`, `declared_queries`, `declared_query_contracts`, `declared_signals`, `declared_signal_contracts`, `declared_updates`, `declared_update_contracts`, `declared_entry_method`, `declared_entry_mode`, `declared_entry_declaring_class`, `parent_workflow_instance_id`, `parent_workflow_run_id`, `parent_sequence`, `workflow_link_id`, `child_call_id`, `retry_policy`, `timeout_policy`, `continued_from_run_id`, `retry_attempt`, `retry_of_child_workflow_run_id`, `timeout_cleanup` (experimental: `allowance_seconds` and the derived `work_deadlines` per timeout kind) | `WorkflowDefinitionFingerprint`, `RunLineageView`, worker history payload consumers | | `WorkflowContinuedAsNew` | `sequence`, `continued_to_run_id`, `continued_to_run_number`, `workflow_link_id`, `closed_reason`, `abandoned_activities` | `WorkflowStepHistory`, `RunLineageView`, `HistoryTimeline`, operator detail projections | | `CooperativeCancellationRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `reason`, `cleanup_deadline_at` | workflow cancellation delivery, cleanup deadline enforcement, history export | +| `WorkflowTimeoutCleanupStarted` | `workflow_command_id`, `timeout_kind` (`execution` or `run`), `work_deadline_at`, `cleanup_deadline_at`, `hard_deadline_at`, `detected_at` | `HistoryTimeline`, timeout cleanup outcome (experimental) | | `CooperativeCancellationDelivered` | `workflow_command_id`, `workflow_run_id`, `sequence`, `call_kind`, `sequence_span`, `operation_sequence`, `operation_sequence_span` | `WorkflowFiberRunner`, `WorkflowStepHistory`, cooperative task bridges, history export | | `ActivityScheduled` | `activity_execution_id`, `activity_class`, `activity_type`, `sequence`, `execution_mode`, `local_activity`, `workflow_task_id`, `activity`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path` | `WorkflowStepHistory`, `WorkflowExecutor`, `QueryStateReplayer`, `ActivityRecovery`, `ParallelChildGroup` | | `ActivityStarted` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `attempt_number`, `execution_mode`, `local_activity`, `workflow_task_id`, `lease_expires_at`, `activity`, `activity_attempt`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path` | `ActivitySnapshot`, `ActivityAttemptSnapshots`, `HistoryTimeline`, `RunActivityView`, `ParallelChildGroup` | @@ -809,12 +810,12 @@ class, source file path, or stack trace to replay or handle a failure. | `MemoUpserted` | `sequence`, `entries`, `merged`, `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `WorkflowExecutor`, `QueryStateReplayer`, `WorkflowFiberRunner`, `HistoryTimeline`, run detail projections, history export, `RedriveReplayClock` | | `RepairRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `outcome`, `liveness_state`, `wait_kind`, `task_id`, `task_type` | `HistoryTimeline`, `RunCommandContract`, repair diagnostics, operator detail projections | | `CancelRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `reason` | `HistoryTimeline`, `RunCommandContract`, cancellation projections | -| `WorkflowCancelled` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason`, `cancellation_cleanup` | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, cancellation projections | +| `WorkflowCancelled` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason`, `cancellation_cleanup`, `timeout_cleanup` (experimental; the deadline-triggered cleanup outcome) | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, cancellation projections | | `TerminateRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `reason` | `HistoryTimeline`, `RunCommandContract`, termination projections | -| `WorkflowTerminated` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason` | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, termination projections | +| `WorkflowTerminated` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason`, `timeout_cleanup` (experimental; the deadline-triggered cleanup outcome) | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, termination projections | | `ArchiveRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `outcome`, `reason` | `HistoryTimeline`, `RunCommandContract`, archive/export projections | | `WorkflowArchived` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `archive_command_id`, `reason` | `HistoryTimeline`, archive/export projections, operator detail projections | -| `WorkflowTimedOut` | `failure_id`, `timeout_kind`, `failure_category`, `message`, `exception_class`, `execution_deadline_at`, `run_deadline_at` | `FailureSnapshots`, `HistoryTimeline`, `ChildRunHistory`, timeout repair projections | +| `WorkflowTimedOut` | `failure_id`, `timeout_kind`, `failure_category`, `message`, `exception_class`, `execution_deadline_at`, `run_deadline_at`, `timeout_cleanup` (experimental; the deadline-triggered cleanup outcome), `abandoned_activities`, `cancellation_cleanup` (preempted by the hard timeout) | `FailureSnapshots`, `HistoryTimeline`, `ChildRunHistory`, timeout repair projections | | `WorkflowCompleted` | `output`, `payload_codec`, `abandoned_activities` | `ChildRunHistory`, `HistoryTimeline`, history export, parent resume projections | | `WorkflowFailed` | `failure_id`, `source_kind`, `source_id`, `failure_category`, `non_retryable`, `exception_type`, `exception_class`, `message`, `exception`, `structural_limit_kind`, `structural_limit_value`, `structural_limit_configured`, `structural_limit_name`, `failed_step_sequence`, `failed_step_kind` (`activity`, `workflow_code` or `structural_limit`), `abandoned_activities` | `FailureSnapshots`, `HistoryTimeline`, `ChildRunHistory`, failure repair projections, `FailedRunRedrivePlan` | | `FailureHandled` | `failure_id`, `sequence`, `failure_category`, `source_kind`, `source_id`, `propagation_kind`, `exception_class`, `exception_type`, `message`, `handled` | `FailureSnapshots`, `HistoryTimeline`, `QueryStateReplayer`, operator detail projections | diff --git a/docs/architecture/timeout-cleanup.md b/docs/architecture/timeout-cleanup.md index 061f285d5..865c2130a 100644 --- a/docs/architecture/timeout-cleanup.md +++ b/docs/architecture/timeout-cleanup.md @@ -46,9 +46,13 @@ Validation at admission, before any write: work deadline is in the future. The allowance is recorded in `WorkflowStarted` (`timeout_cleanup.allowance_seconds`, -plus the derived `work_deadline_at` per timeout kind), which is immutable. There -is no new `workflow_runs` column: the engine and native read the admitted values -from `WorkflowStarted` (server review on [workflow pull request 1122](https://github.com/durable-workflow/workflow/pull/1122)). Describe and run detail show them. +plus the derived `work_deadline_at` per timeout kind), which is immutable. Native +reads the admitted values from `WorkflowStarted` and adds no column (server review +on [workflow pull request 1122](https://github.com/durable-workflow/workflow/pull/1122)). +Embedded also stores the earliest derived work deadline in +`workflow_runs.timeout_cleanup_work_deadline_at`, so the watchdog can wake an idle +run at that time; the column is nullable and set only for opted-in runs (General, +2026-11-13). Describe and run detail show them. ### Inheritance @@ -80,15 +84,20 @@ notices first) does the following in one transaction under the run lock: 3. Delivers the interruption through the existing cooperative-cancellation path. Workflow code receives a `WorkflowTimeoutCleanupRequested` exception, which extends `WorkflowCancellationRequestedException`, with a - `CancellationContext` whose `cause` is `timeout` and whose `timeoutKind`, - `workDeadlineAt` and `cleanupDeadlineAt` are set. Existing + `CancellationContext` whose optional top-level `cause` is `timeout` and + whose existing `cleanup_deadline_at` is the cleanup deadline. The timeout + kind and work deadline are recorded on `WorkflowTimeoutCleanupStarted`, not + in the context. `cause` is never inherited by descendants, scoped or not, + and an explicit `cause: cancellation` round-trips unchanged. Existing `catch (WorkflowCancellationRequestedException)` compensation code therefore runs without changes. `cancellationShield()` scopes and the operation cancellation policies behave as they do for a cooperative cancellation request. A user cancellation request that arrives during timeout cleanup is recorded -but doesn't restart cleanup or move its deadline. A timeout that arrives +as an additional accepted RequestCancellation command referencing the original +engine request. It does not create another interruption, restart cleanup or move +its deadline. A timeout that arrives during an earlier cooperative cancellation doesn't enter timeout cleanup: the earlier request already owns cleanup, its deadline is capped by the hard deadline, and the final outcome is the timeout if the hard deadline wins. @@ -133,6 +142,10 @@ with `cause: timeout`. After that, they show the recorded outcome. new event type, so readers that don't know it refuse it as an unknown event. Service-mode SDKs therefore need the capability below before Server sends it. +- This embedded increment admits the option only with queue dispatch. Poll-mode + starts and service starts refuse timeout cleanup, and bridge claims of an + opted-in embedded run return `timeout_cleanup_not_supported` without taking + the lease. Service support follows in [server issue 1201](https://github.com/durable-workflow/server/issues/1201). - Service mode: Server advertises `timeout_cleanup` (`{supported: true, version: 1}`) only when it implements this contract, and a service worker advertises it only when its SDK delivers the typed timeout interruption. diff --git a/scripts/ci/validate-regression-corpus.py b/scripts/ci/validate-regression-corpus.py index b1a5ba97b..b905c5918 100644 --- a/scripts/ci/validate-regression-corpus.py +++ b/scripts/ci/validate-regression-corpus.py @@ -177,6 +177,7 @@ "tests/Fixtures/V2/TestNexusWaitingCallerWorkflow.php", "tests/Fixtures/V2/TestNexusCancellationCleanupCallerWorkflow.php", "tests/Fixtures/V2/TestNexusDeadlineCallerWorkflow.php", + "tests/Fixtures/V2/TestTimeoutCleanupWorkflow.php", "tests/Support/NexusCallerDeliveryScenario.php", "tests/Unit/V2/ReplayRegressionCorpusTest.php", ), diff --git a/src/V2/CancellationContext.php b/src/V2/CancellationContext.php index d250c8aca..9fe2bac3a 100644 --- a/src/V2/CancellationContext.php +++ b/src/V2/CancellationContext.php @@ -15,6 +15,12 @@ /** Immutable request metadata recorded in canonical workflow history. */ final class CancellationContext { + /** + * Cause of a run's own interruption: `timeout` when deadline-triggered + * cleanup (#863) requested it. Absent means an ordinary cancellation. + */ + public const CAUSE_TIMEOUT = 'timeout'; + /** * @var WeakMap>|null */ @@ -37,9 +43,15 @@ private function __construct( private readonly CarbonImmutable $cleanupDeadline, public readonly array $lineage, public readonly ?ScopedCancellationContext $scopeOrigin = null, + public readonly ?string $cause = null, ) { } + public function isTimeout(): bool + { + return $this->cause === self::CAUSE_TIMEOUT; + } + /** * @param array $snapshot */ @@ -100,6 +112,14 @@ public static function fromArray(array $snapshot): self $rootRunId = self::text($snapshot, 'root_workflow_run_id'); $parentRequestId = $snapshot['parent_request_id'] ?? null; $reason = $snapshot['reason'] ?? null; + // Top-level and optional; absent or `cancellation` is an ordinary + // cancellation (#863). + $cause = $snapshot['cause'] ?? null; + if ($cause !== null && ! in_array($cause, ['cancellation', self::CAUSE_TIMEOUT], true)) { + throw new InvalidArgumentException('Cancellation cause must be cancellation or timeout.'); + } + // An explicit `cancellation` is preserved, so a context round-trips + // with exactly its own keys. if (count(array_unique(array_column($normalizedLineage, 'request_id'))) !== count($normalizedLineage) || count(array_unique(array_column($normalizedLineage, 'workflow_run_id'))) !== count($normalizedLineage)) { throw new InvalidArgumentException('Cancellation lineage cannot contain a cycle.'); @@ -168,6 +188,7 @@ public static function fromArray(array $snapshot): self $deadline, $normalizedLineage, $scopeOrigin, + $cause, ); } @@ -195,6 +216,9 @@ public function forDescendant(string $requestId, string $workflowInstanceId, str ); } $snapshot = $this->toArray(); + // A descendant is cancelled because its ancestor was; it was not + // itself timed out. + unset($snapshot['cause']); $snapshot['request_id'] = $requestId; $snapshot['parent_request_id'] = $this->requestId; $snapshot['lineage'][] = [ @@ -271,6 +295,9 @@ public function toArray(): array 'requested_at' => $this->originalRequestedAt->toISOString(), 'cleanup_deadline_at' => $this->cleanupDeadline->toISOString(), 'lineage' => $this->lineage, + ...($this->cause === null ? [] : [ + 'cause' => $this->cause, + ]), ...($this->scopeOrigin === null ? [] : [ 'scope_origin' => $this->scopeOrigin->toArray(), 'scope_authority_deadline_at' => $this->cleanupDeadline->toISOString(), @@ -313,6 +340,10 @@ private static function scopedDescendantSnapshot( $lineage[] = $address; } } + // A descendant is cancelled because its ancestor was; it was not + // itself timed out (#863). + unset($root['cause']); + return [ ...$root, 'schema' => 'durable-workflow.cancellation-context/v2', diff --git a/src/V2/CommandContext.php b/src/V2/CommandContext.php index ad267b401..cae9baa97 100644 --- a/src/V2/CommandContext.php +++ b/src/V2/CommandContext.php @@ -31,6 +31,26 @@ public static function phpApi(): self ]); } + /** + * A command the engine issues itself, such as entering timeout cleanup at + * a run's work deadline (#863). + * + * @internal Engine attribution, not an application command context. + */ + public static function system(string $id): self + { + return new self('system', [ + 'caller' => [ + 'type' => 'system', + 'label' => $id, + ], + 'auth' => [ + 'status' => 'not_applicable', + 'method' => 'none', + ], + ]); + } + public static function controlPlane(): self { return new self('control_plane', [ diff --git a/src/V2/Enums/HistoryEventType.php b/src/V2/Enums/HistoryEventType.php index 4594e418b..dbf2141aa 100644 --- a/src/V2/Enums/HistoryEventType.php +++ b/src/V2/Enums/HistoryEventType.php @@ -71,6 +71,8 @@ enum HistoryEventType: string case SearchAttributesUpserted = 'SearchAttributesUpserted'; case MemoUpserted = 'MemoUpserted'; case WorkflowTimedOut = 'WorkflowTimedOut'; + + case WorkflowTimeoutCleanupStarted = 'WorkflowTimeoutCleanupStarted'; case WorkflowCompleted = 'WorkflowCompleted'; case WorkflowFailed = 'WorkflowFailed'; case ParentClosePolicyApplied = 'ParentClosePolicyApplied'; diff --git a/src/V2/Exceptions/WorkflowCancellationRequestedException.php b/src/V2/Exceptions/WorkflowCancellationRequestedException.php index a1353e77b..243c91d00 100644 --- a/src/V2/Exceptions/WorkflowCancellationRequestedException.php +++ b/src/V2/Exceptions/WorkflowCancellationRequestedException.php @@ -8,7 +8,7 @@ use Throwable; use Workflow\V2\CancellationContext; -final class WorkflowCancellationRequestedException extends Error +class WorkflowCancellationRequestedException extends Error { public function __construct( string $message = 'Cooperative cancellation requested.', @@ -18,4 +18,15 @@ public function __construct( ) { parent::__construct($message, $code, $previous); } + + /** + * The exception for a delivered request: the timeout cleanup subtype when + * deadline-triggered cleanup (#863) requested it. + */ + public static function forContext(?CancellationContext $cancellation): self + { + return $cancellation?->isTimeout() === true + ? new WorkflowTimeoutCleanupRequested(cancellation: $cancellation) + : new self(cancellation: $cancellation); + } } diff --git a/src/V2/Exceptions/WorkflowTimeoutCleanupRequested.php b/src/V2/Exceptions/WorkflowTimeoutCleanupRequested.php new file mode 100644 index 000000000..9e410fae4 --- /dev/null +++ b/src/V2/Exceptions/WorkflowTimeoutCleanupRequested.php @@ -0,0 +1,25 @@ + 'integer', 'execution_deadline_at' => 'datetime', 'run_deadline_at' => 'datetime', + 'timeout_cleanup_work_deadline_at' => 'datetime', 'cancellation_requested_at' => 'datetime', 'cancellation_deadline_at' => 'datetime', 'cancellation_delivered_at' => 'datetime', diff --git a/src/V2/StartOptions.php b/src/V2/StartOptions.php index 9ddc1ed39..9883586ae 100644 --- a/src/V2/StartOptions.php +++ b/src/V2/StartOptions.php @@ -42,6 +42,15 @@ final class StartOptions */ public readonly ?int $runTimeoutSeconds; + /** + * Seconds reserved at the end of the hard execution and run timeouts for + * workflow-authored cleanup (#863). When the work deadline (hard deadline + * minus this allowance) passes, workflow code receives a timeout cleanup + * interruption; the run still ends timed out at the hard deadline at the + * latest. Null keeps today's hard timeout. Experimental. + */ + public readonly ?int $timeoutCleanupSeconds; + /** * Dispatch priority for the workflow's tasks. Lower numbers run first * when workers on a shared task queue are saturated. @@ -77,6 +86,7 @@ public function __construct( ?int $priority = null, ?string $fairnessKey = null, ?int $fairnessWeight = null, + ?int $timeoutCleanupSeconds = null, ) { $this->duplicateStartPolicy = $duplicateStartPolicy; $this->businessKey = self::normalizeBusinessKey($businessKey); @@ -88,6 +98,11 @@ public function __construct( $this->priority = TaskPriority::normalize($priority); $this->fairnessKey = TaskFairnessKey::normalize($fairnessKey); $this->fairnessWeight = TaskFairnessKey::normalizeWeight($fairnessWeight); + $this->timeoutCleanupSeconds = self::normalizeTimeoutCleanup( + $timeoutCleanupSeconds, + $this->executionTimeoutSeconds, + $this->runTimeoutSeconds, + ); } public static function rejectDuplicate(): self @@ -162,6 +177,16 @@ public function withRunTimeout(?int $seconds): self ]); } + /** + * Reserve the last `$seconds` of the hard timeouts for cleanup (#863). + */ + public function withTimeoutCleanup(?int $seconds): self + { + return $this->cloneWith([ + 'timeoutCleanupSeconds' => $seconds, + ]); + } + public function withPriority(?int $priority): self { return $this->cloneWith([ @@ -199,6 +224,10 @@ private function cloneWith(array $overrides): self array_key_exists('priority', $overrides) ? $overrides['priority'] : $this->priority, array_key_exists('fairnessKey', $overrides) ? $overrides['fairnessKey'] : $this->fairnessKey, array_key_exists('fairnessWeight', $overrides) ? $overrides['fairnessWeight'] : $this->fairnessWeight, + array_key_exists( + 'timeoutCleanupSeconds', + $overrides + ) ? $overrides['timeoutCleanupSeconds'] : $this->timeoutCleanupSeconds, ); } @@ -404,6 +433,46 @@ private static function normalizeSearchAttributes(array $searchAttributes): arra return $normalized; } + /** + * The cleanup allowance must leave a work deadline in the future for every + * hard timeout it applies to; it never extends a hard timeout. + */ + private static function normalizeTimeoutCleanup( + ?int $seconds, + ?int $executionTimeoutSeconds, + ?int $runTimeoutSeconds, + ): ?int { + if ($seconds === null) { + return null; + } + + if ($seconds < 1 || $seconds > 86400) { + throw new LogicException('Workflow v2 timeout cleanup must be between 1 and 86400 seconds.'); + } + + if ($executionTimeoutSeconds === null && $runTimeoutSeconds === null) { + throw new LogicException( + 'Workflow v2 timeout cleanup needs an execution or run timeout to reserve time from.' + ); + } + + foreach ([ + 'execution' => $executionTimeoutSeconds, + 'run' => $runTimeoutSeconds, + ] as $kind => $timeout) { + if ($timeout !== null && $seconds >= $timeout) { + throw new LogicException(sprintf( + 'Workflow v2 timeout cleanup (%d s) must be shorter than the %s timeout (%d s).', + $seconds, + $kind, + $timeout, + )); + } + } + + return $seconds; + } + private static function normalizeTimeout(?int $seconds, string $kind): ?int { if ($seconds === null) { diff --git a/src/V2/Support/ActivityAttemptAuthority.php b/src/V2/Support/ActivityAttemptAuthority.php index 337529b96..4d275aa5a 100644 --- a/src/V2/Support/ActivityAttemptAuthority.php +++ b/src/V2/Support/ActivityAttemptAuthority.php @@ -35,6 +35,12 @@ private static function expiredBusinessReason( return $kind . '_timeout'; } + if ($run !== null && ClosingRunOpenWork::abandonedAtClose($run, $execution)) { + // The terminal event explicitly retained this independent + // lifetime. Its own deadlines and lease still fence the attempt. + return null; + } + foreach ([ 'run_timeout' => $run?->run_deadline_at, 'execution_timeout' => $run?->execution_deadline_at, diff --git a/src/V2/Support/ClosingRunOpenWork.php b/src/V2/Support/ClosingRunOpenWork.php index 9fd0e734f..ecb3af62e 100644 --- a/src/V2/Support/ClosingRunOpenWork.php +++ b/src/V2/Support/ClosingRunOpenWork.php @@ -135,7 +135,7 @@ public static function settle(WorkflowRun $run, WorkflowTask $closingTask): arra */ public static function abandonedAtClose(WorkflowRun $run, ActivityExecution $execution): bool { - if (! $run->status->isTerminal() || $execution->workflow_run_id !== $run->id + if ($run->status?->isTerminal() !== true || $execution->workflow_run_id !== $run->id || ! self::abandonsAtClose($execution)) { return false; } @@ -147,6 +147,7 @@ public static function abandonedAtClose(WorkflowRun $run, ActivityExecution $exe HistoryEventType::WorkflowCompleted->value, HistoryEventType::WorkflowFailed->value, HistoryEventType::WorkflowContinuedAsNew->value, + HistoryEventType::WorkflowTimedOut->value, ]) ->orderByDesc('sequence') ->first(); diff --git a/src/V2/Support/DefaultWorkflowControlPlane.php b/src/V2/Support/DefaultWorkflowControlPlane.php index e6abf71d5..3d9f160d9 100644 --- a/src/V2/Support/DefaultWorkflowControlPlane.php +++ b/src/V2/Support/DefaultWorkflowControlPlane.php @@ -189,6 +189,15 @@ public function redrive(string $instanceId, string $failedRunId, array $options 'run_timeout_seconds' => $runTimeoutSeconds, 'execution_deadline_at' => $executionDeadlineAt, 'run_deadline_at' => $runDeadlineAt, + // Only an opted-in run writes the column, so custom run tables without + // it keep working until they migrate (#863). + ...(($timeoutCleanupWorkDeadline = TimeoutCleanup::workDeadline( + $timeoutCleanupAllowance = TimeoutCleanup::allowanceForRun($source), + $executionDeadlineAt, + $runDeadlineAt, + )) === null ? [] : [ + 'timeout_cleanup_work_deadline_at' => $timeoutCleanupWorkDeadline, + ]), 'connection' => $source->connection, 'queue' => $source->queue, 'priority' => $source->priority, @@ -250,6 +259,13 @@ public function redrive(string $instanceId, string $failedRunId, array $options 'recovery_kind' => 'redrive', 'execution_deadline_at' => $executionDeadlineAt?->toIso8601String(), 'run_deadline_at' => $runDeadlineAt?->toIso8601String(), + ...(($timeoutCleanup = TimeoutCleanup::startedPayload( + $timeoutCleanupAllowance, + $executionDeadlineAt, + $runDeadlineAt, + )) === null ? [] : [ + TimeoutCleanup::STARTED_KEY => $timeoutCleanup, + ]), 'replayed_started_at' => is_string($startedPayload['replayed_started_at'] ?? null) ? $startedPayload['replayed_started_at'] : $source->started_at?->toIso8601String(), @@ -308,6 +324,17 @@ public function redrive(string $instanceId, string $failedRunId, array $options public function start(string $workflowType, ?string $instanceId = null, array $options = []): array { + if (($options['timeout_cleanup_seconds'] ?? null) !== null) { + return [ + 'accepted' => false, + 'workflow_instance_id' => $instanceId, + 'workflow_run_id' => null, + 'task_id' => null, + 'reason' => 'timeout_cleanup_not_supported', + 'status' => 422, + ]; + } + $resolvedClass = $this->tryResolveWorkflowClass($workflowType); $arguments = $options['arguments'] ?? null; $payloadCodec = isset($options['payload_codec']) && is_string( diff --git a/src/V2/Support/DefaultWorkflowTaskBridge.php b/src/V2/Support/DefaultWorkflowTaskBridge.php index 0d2b65720..e59f6d03a 100644 --- a/src/V2/Support/DefaultWorkflowTaskBridge.php +++ b/src/V2/Support/DefaultWorkflowTaskBridge.php @@ -294,6 +294,16 @@ public function claimStatus(string $taskId, ?string $leaseOwner = null): array return self::claimRejected($taskId, 'run_closed', "The workflow run is {$run->status->value}."); } + // This increment admits timeout cleanup only for embedded queue + // workers. A bridge worker must not silently run ordinary work. + if ($run->timeout_cleanup_work_deadline_at !== null) { + return self::claimRejected( + $taskId, + 'timeout_cleanup_not_supported', + 'Timeout cleanup requires embedded queue workers.' + ); + } + TaskCompatibility::sync($task, $run); if (TaskBackendCapabilities::recordClaimFailureIfUnsupported($task) !== null) { @@ -5349,6 +5359,15 @@ private function applyContinueAsNew( 'run_timeout_seconds' => $runTimeoutSeconds, 'execution_deadline_at' => $executionDeadlineAt, 'run_deadline_at' => $runDeadlineAt, + // Only an opted-in run writes the column, so custom run tables without + // it keep working until they migrate (#863). + ...(($timeoutCleanupWorkDeadline = TimeoutCleanup::workDeadline( + $timeoutCleanupAllowance = TimeoutCleanup::allowanceForRun($run), + $executionDeadlineAt, + $runDeadlineAt, + )) === null ? [] : [ + 'timeout_cleanup_work_deadline_at' => $timeoutCleanupWorkDeadline, + ]), 'connection' => $run->connection, 'queue' => $queue, ...TaskSchedulingFields::nonDefault(TaskSchedulingFields::forRun($run)), @@ -5491,6 +5510,13 @@ private function applyContinueAsNew( 'workflow_type' => $workflowType, 'workflow_instance_id' => $continuedRun->workflow_instance_id, 'workflow_run_id' => $continuedRun->id, + ...(($timeoutCleanup = TimeoutCleanup::startedPayload( + $timeoutCleanupAllowance, + $executionDeadlineAt, + $runDeadlineAt, + )) === null ? [] : [ + TimeoutCleanup::STARTED_KEY => $timeoutCleanup, + ]), 'continued_from_run_id' => $run->id, 'workflow_link_id' => $link->id, 'child_call_id' => $childCallId, diff --git a/src/V2/Support/HistoryEventPayloadContract.php b/src/V2/Support/HistoryEventPayloadContract.php index db011a51a..c33b856ae 100644 --- a/src/V2/Support/HistoryEventPayloadContract.php +++ b/src/V2/Support/HistoryEventPayloadContract.php @@ -127,6 +127,7 @@ final class HistoryEventPayloadContract 'replayed_started_at', 'retry_attempt', 'retry_of_child_workflow_run_id', + 'timeout_cleanup', ], 'WorkflowContinuedAsNew' => [ 'sequence', @@ -990,6 +991,7 @@ final class HistoryEventPayloadContract 'message', 'reason', 'cancellation_cleanup', + 'timeout_cleanup', ], 'TerminateRequested' => [ 'workflow_command_id', @@ -1008,6 +1010,7 @@ final class HistoryEventPayloadContract 'exception_class', 'message', 'reason', + 'timeout_cleanup', ], 'ArchiveRequested' => [ 'workflow_command_id', @@ -1032,6 +1035,17 @@ final class HistoryEventPayloadContract 'exception_class', 'execution_deadline_at', 'run_deadline_at', + 'timeout_cleanup', + 'abandoned_activities', + 'cancellation_cleanup', + ], + 'WorkflowTimeoutCleanupStarted' => [ + 'workflow_command_id', + 'timeout_kind', + 'work_deadline_at', + 'cleanup_deadline_at', + 'hard_deadline_at', + 'detected_at', ], 'WorkflowCompleted' => ['output', 'payload_codec', 'abandoned_activities'], 'WorkflowFailed' => [ diff --git a/src/V2/Support/HistoryTimeline.php b/src/V2/Support/HistoryTimeline.php index 359edfcaf..e2dff29be 100644 --- a/src/V2/Support/HistoryTimeline.php +++ b/src/V2/Support/HistoryTimeline.php @@ -618,6 +618,10 @@ private static function summaryFor( }, HistoryEventType::CancelRequested => 'Cancel requested.', HistoryEventType::CooperativeCancellationRequested => 'Cooperative cancellation requested.', + HistoryEventType::WorkflowTimeoutCleanupStarted => sprintf( + 'Work deadline passed; timeout cleanup runs until %s.', + self::stringValue($payload['cleanup_deadline_at'] ?? null) ?? 'the hard deadline', + ), HistoryEventType::CooperativeCancellationDelivered => 'Cancellation delivered to workflow code.', HistoryEventType::WorkflowCancelled => match ($payload['cancellation_cleanup']['outcome'] ?? null) { 'completed' => 'Workflow cancelled after cleanup completed.', @@ -729,13 +733,13 @@ private static function summaryFor( self::intValue($payload['new_position'] ?? null) ?? 0, ), HistoryEventType::WorkflowCompleted => 'Workflow completed.', - HistoryEventType::WorkflowTimedOut => match ($payload['timeout_kind'] ?? null) { + HistoryEventType::WorkflowTimedOut => trim(match ($payload['timeout_kind'] ?? null) { 'execution_timeout' => 'Workflow execution deadline expired.', 'run_timeout' => 'Workflow run deadline expired.', default => $message === null ? 'Workflow timed out.' : sprintf('Workflow timed out: %s.', $message), - }, + } . ' ' . (TimeoutCleanup::summary($payload[TimeoutCleanup::STARTED_KEY] ?? null) ?? '')), HistoryEventType::WorkflowFailed => $message === null ? 'Workflow failed.' : sprintf('Workflow failed: %s.', $message), @@ -1396,6 +1400,7 @@ private static function historicalCommandOutcome( : 'update_completed', HistoryEventType::CancelRequested => 'cancelled', HistoryEventType::CooperativeCancellationRequested => 'cancellation_requested', + HistoryEventType::WorkflowTimeoutCleanupStarted => 'cancellation_requested', HistoryEventType::TerminateRequested => 'terminated', HistoryEventType::ArchiveRequested => 'archived', default => $command?->outcome?->value, diff --git a/src/V2/Support/TimeoutCleanup.php b/src/V2/Support/TimeoutCleanup.php new file mode 100644 index 000000000..d541f916e --- /dev/null +++ b/src/V2/Support/TimeoutCleanup.php @@ -0,0 +1,296 @@ +timeout_cleanup_work_deadline_at; + if ($workDeadline === null || $run->status->isTerminal() + || is_string($run->cancellation_request_command_id) || $now->lt($workDeadline)) { + return false; + } + + [$timeoutKind, $hardDeadline] = self::earliestHardDeadline($run); + if ($hardDeadline === null || $now->gte($hardDeadline)) { + // The hard timeout closes the run instead. + return false; + } + + $run->loadMissing(['instance', 'historyEvents']); + if ($run->historyEvents->contains( + static fn (WorkflowHistoryEvent $event): bool => $event->event_type === HistoryEventType::WorkflowTimeoutCleanupStarted + )) { + return false; + } + + /** @var WorkflowInstance $instance */ + $instance = $run->instance; + $requestId = (string) Str::ulid(); + $reason = 'Workflow work deadline passed; timeout cleanup requested.'; + $context = [ + 'schema' => 'durable-workflow.cancellation-context/v1', + 'request_id' => $requestId, + 'root_request_id' => $requestId, + 'root_workflow_instance_id' => $instance->id, + 'root_workflow_run_id' => $run->id, + 'parent_request_id' => null, + 'reason' => $reason, + 'requester' => [ + 'type' => 'system', + 'id' => 'timeout_cleanup', + ], + 'source' => 'system', + 'requested_at' => $now->toISOString(), + 'cleanup_deadline_at' => $hardDeadline->toISOString(), + 'lineage' => [[ + 'request_id' => $requestId, + 'workflow_instance_id' => $instance->id, + 'workflow_run_id' => $run->id, + ]], + 'cause' => CancellationContext::CAUSE_TIMEOUT, + ]; + $context = CancellationContext::fromArray($context)->toArray(); + $codec = $run->payload_codec ?? CodecRegistry::defaultCodec(); + + $command = WorkflowCommand::record($instance, $run, [ + 'id' => $requestId, + 'command_type' => CommandType::RequestCancellation->value, + ...CommandContext::system('timeout_cleanup')->attributes(), + 'target_scope' => 'run', + 'status' => CommandStatus::Accepted->value, + 'outcome' => CommandOutcome::CancellationRequested->value, + 'payload_codec' => $codec, + 'payload' => Serializer::serializeWithCodec($codec, [ + 'reason' => $reason, + 'cleanup_deadline_at' => $hardDeadline->toISOString(), + 'cancellation' => $context, + ]), + 'accepted_at' => $now, + 'applied_at' => $now, + ]); + + $run->forceFill([ + 'cancellation_request_command_id' => $command->id, + 'cancellation_requested_at' => $now, + 'cancellation_deadline_at' => $hardDeadline, + 'last_progress_at' => $now, + ])->save(); + + WorkflowHistoryEvent::record($run, HistoryEventType::WorkflowTimeoutCleanupStarted, [ + 'workflow_command_id' => $command->id, + 'timeout_kind' => $timeoutKind, + 'work_deadline_at' => $workDeadline->toIso8601String(), + 'cleanup_deadline_at' => $hardDeadline->toIso8601String(), + 'hard_deadline_at' => $hardDeadline->toIso8601String(), + 'detected_at' => $now->toIso8601String(), + ], null, $command); + WorkflowHistoryEvent::record($run, HistoryEventType::CooperativeCancellationRequested, [ + 'workflow_command_id' => $command->id, + 'workflow_instance_id' => $instance->id, + 'workflow_run_id' => $run->id, + 'command_type' => CommandType::RequestCancellation->value, + 'reason' => $reason, + 'cleanup_deadline_at' => $hardDeadline->toISOString(), + 'cancellation' => $context, + ], null, $command); + + CancellationScopeDescendants::request($run, CancellationScopeHistory::ROOT_SCOPE_ID); + + $run->unsetRelation('historyEvents'); + $run->unsetRelation('commands'); + + return true; + } + + /** + * The `timeout_cleanup` outcome recorded on the terminal event. + * + * @param 'completed'|'failed'|'expired'|'preempted' $outcome + * @return array + */ + public static function outcome(WorkflowRun $run, string $outcome, ?string $failureId = null): array + { + /** @var WorkflowHistoryEvent|null $started */ + $started = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::WorkflowTimeoutCleanupStarted->value) + ->first(); + $payload = is_array($started?->payload) ? $started->payload : []; + [$kind, $hardDeadline] = self::earliestHardDeadline($run); + + return array_filter([ + 'outcome' => $outcome, + 'timeout_kind' => $payload['timeout_kind'] ?? $kind, + 'started_at' => $payload['detected_at'] ?? null, + 'cleanup_deadline_at' => $payload['cleanup_deadline_at'] ?? $hardDeadline?->toIso8601String(), + 'finished_at' => now() + ->toIso8601String(), + 'failure_id' => $failureId, + ], static fn (mixed $value): bool => $value !== null); + } + + /** + * A plain sentence for an outcome, so describe, the export, the failure + * list and the timeline never present a failed cleanup as a clean + * timeout (#863). + */ + public static function summary(mixed $outcome): ?string + { + return match (is_array($outcome) ? ($outcome['outcome'] ?? null) : null) { + 'completed' => 'Timeout cleanup completed.', + 'failed' => 'Timeout cleanup failed.', + 'expired' => 'Timeout cleanup did not finish before the hard deadline.', + 'preempted' => 'Timeout cleanup was preempted.', + default => null, + }; + } + + /** + * @return array{0: string|null, 1: CarbonInterface|null} + */ + public static function earliestHardDeadline(WorkflowRun $run): array + { + $kind = null; + $earliest = null; + foreach ([ + 'execution' => $run->execution_deadline_at, + 'run' => $run->run_deadline_at, + ] as $candidateKind => $deadline) { + if ($deadline !== null && ($earliest === null || $deadline->lt($earliest))) { + $kind = $candidateKind; + $earliest = $deadline; + } + } + + return [$kind, $earliest]; + } + + /** + * The allowance a run was admitted with, read from its `WorkflowStarted`. + * A continued or redriven run inherits it, as it inherits the hard + * deadlines. + */ + public static function allowanceForRun(WorkflowRun $run): ?int + { + /** @var WorkflowHistoryEvent|null $started */ + $started = $run->relationLoaded('historyEvents') + ? $run->historyEvents->first( + static fn (WorkflowHistoryEvent $event): bool => $event->event_type === HistoryEventType::WorkflowStarted + ) + : $run->historyEvents() + ->where('event_type', HistoryEventType::WorkflowStarted->value) + ->first(); + $allowance = $started?->payload[self::STARTED_KEY]['allowance_seconds'] ?? null; + + return is_int($allowance) && $allowance > 0 ? $allowance : null; + } + + /** + * The earliest work deadline: each hard deadline minus the allowance. + */ + public static function workDeadline( + ?int $allowanceSeconds, + ?CarbonInterface $executionDeadlineAt, + ?CarbonInterface $runDeadlineAt, + ): ?CarbonInterface { + if ($allowanceSeconds === null) { + return null; + } + + $earliest = null; + foreach ([$executionDeadlineAt, $runDeadlineAt] as $hardDeadline) { + if ($hardDeadline === null) { + continue; + } + $work = $hardDeadline->copy() + ->subSeconds($allowanceSeconds); + if ($earliest === null || $work->lt($earliest)) { + $earliest = $work; + } + } + + return $earliest; + } + + /** + * @return array{allowance_seconds: int, work_deadlines: array}|null + */ + public static function startedPayload( + ?int $allowanceSeconds, + ?CarbonInterface $executionDeadlineAt, + ?CarbonInterface $runDeadlineAt, + ): ?array { + if ($allowanceSeconds === null) { + return null; + } + + $workDeadlines = []; + foreach ([ + 'execution' => $executionDeadlineAt, + 'run' => $runDeadlineAt, + ] as $kind => $hardDeadline) { + if ($hardDeadline !== null) { + $workDeadlines[$kind] = $hardDeadline->copy() + ->subSeconds($allowanceSeconds) + ->toIso8601String(); + } + } + + return [ + 'allowance_seconds' => $allowanceSeconds, + 'work_deadlines' => $workDeadlines, + ]; + } +} diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index 4b0282248..a2a78ff84 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -161,6 +161,11 @@ public function cancelIfCleanupDeadlineExpired(WorkflowRun $run, WorkflowTask $t 'childLinks.childRun.failures', 'childLinks.childRun.historyEvents', ]); + if ($this->deadlineExpired($run)) { + $this->timeoutRun($run, $task); + + return true; + } $this->finishCooperativeCancellation($run, $task, 'Cooperative cancellation cleanup deadline expired.'); return true; @@ -278,6 +283,12 @@ private function runTask( return null; } + // At the work deadline an opted-in run enters timeout cleanup once and + // this task runs workflow code with the interruption (#863). + if (TimeoutCleanup::enter($run, now())) { + $run->load('historyEvents'); + } + $sample = WorkflowExecutionProfile::begin('initialization'); try { $state = $this->prepareExecution($run, $task, $continuation); @@ -2469,9 +2480,7 @@ private function resumeCancellation( try { $this->syncWorkflowCursor($frame->workflow, $frame->sequence + $size); $frame->current = $frame->execution->throw( - new WorkflowCancellationRequestedException( - cancellation: CooperativeCancellationDelivery::context($run), - ), + WorkflowCancellationRequestedException::forContext(CooperativeCancellationDelivery::context($run)), $run->cancellation_delivered_at, ); } catch (Throwable $throwable) { @@ -3745,6 +3754,15 @@ private function continueAsNew( 'run_timeout_seconds' => $runTimeoutSeconds, 'execution_deadline_at' => $executionDeadlineAt, 'run_deadline_at' => $runDeadlineAt, + // Only an opted-in run writes the column, so custom run tables without + // it keep working until they migrate (#863). + ...(($timeoutCleanupWorkDeadline = TimeoutCleanup::workDeadline( + $timeoutCleanupAllowance = TimeoutCleanup::allowanceForRun($run), + $executionDeadlineAt, + $runDeadlineAt, + )) === null ? [] : [ + 'timeout_cleanup_work_deadline_at' => $timeoutCleanupWorkDeadline, + ]), 'status' => RunStatus::Pending->value, 'compatibility' => $run->compatibility, 'payload_codec' => $run->payload_codec, @@ -3930,6 +3948,13 @@ private function continueAsNew( 'memo' => $continuedMemo, 'search_attributes' => $continuedSearchAttributes, 'workflow_definition_fingerprint' => WorkflowDefinition::fingerprint($continuedRun->workflow_class), + ...(($timeoutCleanup = TimeoutCleanup::startedPayload( + $timeoutCleanupAllowance, + $executionDeadlineAt, + $runDeadlineAt, + )) === null ? [] : [ + TimeoutCleanup::STARTED_KEY => $timeoutCleanup, + ]), 'continued_from_run_id' => $run->id, 'workflow_link_id' => $link->id, 'child_call_id' => $childCallId, @@ -4008,6 +4033,16 @@ private function finishCooperativeCancellation(WorkflowRun $run, WorkflowTask $t throw new LogicException('Cooperative cancellation has no request command.'); } + // Deadline-triggered cleanup (#863) always ends timed out; the cleanup + // outcome is reported beside it. + if (CooperativeCancellationDelivery::context($run)?->isTimeout() === true) { + $expired = $run->cancellation_deadline_at !== null && now() + ->gte($run->cancellation_deadline_at); + $this->timeoutRun($run, $task, TimeoutCleanup::outcome($run, $expired ? 'expired' : 'completed')); + + return; + } + // The caller holds the run lock; acquiring the instance lock here would invert command lock order. $preservedActivityIds = $run->activityExecutions ->filter(static fn (ActivityExecution $execution): bool => in_array( @@ -4271,23 +4306,52 @@ private function deadlineExpired(WorkflowRun $run): bool return false; } - private function timeoutRun(WorkflowRun $run, WorkflowTask $task): void + /** + * @param array|null $timeoutCleanup The cleanup outcome when + * deadline-triggered cleanup (#863) ran first; its timeout kind wins. + */ + private function timeoutRun(WorkflowRun $run, WorkflowTask $task, ?array $timeoutCleanup = null): void { $now = now(); - $exception = ($run->execution_deadline_at !== null && $now->gte($run->execution_deadline_at)) + // Every deadline boundary, including task completion, preserves the + // cleanup outcome. The watchdog is only one way to notice expiry. + if ($timeoutCleanup === null && (CooperativeCancellationDelivery::context($run)?->isTimeout() === true + || ($run->timeout_cleanup_work_deadline_at !== null && $run->cancellation_request_command_id === null))) { + $timeoutCleanup = TimeoutCleanup::outcome($run, 'expired'); + } + + $cleanupKind = $timeoutCleanup['timeout_kind'] ?? null; + $executionTimedOut = $cleanupKind !== null + ? $cleanupKind === 'execution' + : ($run->execution_deadline_at !== null && $now->gte($run->execution_deadline_at)); + $exception = $executionTimedOut ? WorkflowTimeoutException::executionTimeout($run->execution_deadline_at->toIso8601String()) : WorkflowTimeoutException::runTimeout($run->run_deadline_at->toIso8601String()); $timeoutKind = $exception->timeoutKind; $message = $exception->getMessage(); + $cleanupSummary = TimeoutCleanup::summary($timeoutCleanup); + if ($cleanupSummary !== null) { + // The failure row's message is what describe, the export and the + // failure list show (#863). + $message .= ' ' . $cleanupSummary; + } $exceptionClass = WorkflowTimeoutException::class; + $abandonedActivities = $timeoutCleanup !== null ? ClosingRunOpenWork::settle($run, $task) : []; + $preservedActivityIds = array_column($abandonedActivities, 'activity_execution_id'); + if ($timeoutCleanup !== null) { + $run->load(['tasks', 'activityExecutions', 'timers']); + } + // Cancel all open tasks except the current one. $openTasks = $run->tasks ->filter( static fn (WorkflowTask $t): bool => in_array($t->status, [TaskStatus::Ready, TaskStatus::Leased], true) && $t->id !== $task->id + && ! ($t->task_type === TaskType::Activity + && in_array($t->payload['activity_execution_id'] ?? null, $preservedActivityIds, true)) ); foreach ($openTasks as $openTask) { @@ -4308,7 +4372,7 @@ private function timeoutRun(WorkflowRun $run, WorkflowTask $task): void static fn (ActivityExecution $e): bool => in_array($e->status, [ ActivityStatus::Pending, ActivityStatus::Running, - ], true) + ], true) && ! in_array($e->id, $preservedActivityIds, true) ); foreach ($openActivityExecutions as $execution) { @@ -4370,6 +4434,20 @@ private function timeoutRun(WorkflowRun $run, WorkflowTask $task): void 'exception_class' => $exceptionClass, 'execution_deadline_at' => $run->execution_deadline_at?->toIso8601String(), 'run_deadline_at' => $run->run_deadline_at?->toIso8601String(), + ...($timeoutCleanup === null ? [] : [ + TimeoutCleanup::STARTED_KEY => $timeoutCleanup, + ]), + ...($abandonedActivities === [] ? [] : [ + ClosingRunOpenWork::ABANDONED_ACTIVITIES => $abandonedActivities, + ]), + ...($timeoutCleanup === null && is_string($run->cancellation_request_command_id) ? [ + 'cancellation_cleanup' => [ + 'request_id' => $run->cancellation_request_command_id, + 'outcome' => 'preempted', + 'cleanup_deadline_at' => $run->cancellation_deadline_at?->toISOString(), + 'finished_at' => $now->toISOString(), + ], + ] : []), ], $task); PendingUpdateCloser::closeForTerminalRun($run, $task); @@ -4457,6 +4535,29 @@ private function failRun( return; } + if (is_string($run->cancellation_request_command_id) + && ! $throwable instanceof WorkflowCancellationRequestedException + && CooperativeCancellationDelivery::context($run)?->isTimeout() === true) { + // Timeout cleanup threw: the run is still timed out; the cleanup + // failure is recorded and linked (#863). + /** @var WorkflowFailure $cleanupFailure */ + $cleanupFailure = WorkflowFailure::query()->create(array_merge( + FailureFactory::make($throwable), + [ + 'workflow_run_id' => $run->id, + 'source_kind' => $sourceKind, + 'source_id' => $sourceId, + 'propagation_kind' => 'timeout_cleanup', + 'failure_category' => FailureFactory::classify('terminal', $sourceKind, $throwable)->value, + 'non_retryable' => true, + 'handled' => false, + ], + )); + $this->timeoutRun($run, $task, TimeoutCleanup::outcome($run, 'failed', (string) $cleanupFailure->id)); + + return; + } + $failureCategory = FailureFactory::classify('terminal', $sourceKind, $throwable); $nonRetryable = FailureFactory::isNonRetryable($throwable); diff --git a/src/V2/Support/WorkflowFiberRunner.php b/src/V2/Support/WorkflowFiberRunner.php index cc7606440..1ea52a870 100644 --- a/src/V2/Support/WorkflowFiberRunner.php +++ b/src/V2/Support/WorkflowFiberRunner.php @@ -351,7 +351,7 @@ private function nextObservableStep(): WorkflowStep $this->sequence += $current instanceof AllCall ? $current->leafCount() : 1; $this->throwIntoWorkflow( - new WorkflowCancellationRequestedException(cancellation: $cancellation['context']), + WorkflowCancellationRequestedException::forContext($cancellation['context']), $cancellation['recorded_at'], ); diff --git a/src/V2/TaskWatchdog.php b/src/V2/TaskWatchdog.php index b9fa5e197..757b647a5 100644 --- a/src/V2/TaskWatchdog.php +++ b/src/V2/TaskWatchdog.php @@ -39,6 +39,7 @@ use Workflow\V2\Support\TaskRepair; use Workflow\V2\Support\TaskRepairCandidates; use Workflow\V2\Support\TaskRepairPolicy; +use Workflow\V2\Support\TimeoutCleanup; use Workflow\V2\Support\WorkerCompatibilityFleet; use Workflow\V2\Support\WorkflowExecutor; @@ -664,6 +665,11 @@ private static function deadlineExpiredRunQuery( $cancellation->whereNotNull('cancellation_request_command_id') ->whereNotNull('cancellation_deadline_at') ->where('cancellation_deadline_at', '<=', $now); + })->orWhere(static function ($cleanup) use ($now): void { + // An opted-in run reaches its work deadline (#863). + $cleanup->whereNull('cancellation_request_command_id') + ->whereNotNull('timeout_cleanup_work_deadline_at') + ->where('timeout_cleanup_work_deadline_at', '<=', $now); }); }) ->where(static function ($available) use ($now): void { @@ -674,6 +680,10 @@ private static function deadlineExpiredRunQuery( $cancellation->whereNotNull('cancellation_request_command_id') ->whereNotNull('cancellation_deadline_at') ->where('cancellation_deadline_at', '<=', $now); + })->orWhere(static function ($cleanup) use ($now): void { + $cleanup->whereNull('cancellation_request_command_id') + ->whereNotNull('timeout_cleanup_work_deadline_at') + ->where('timeout_cleanup_work_deadline_at', '<=', $now); }); }); @@ -756,15 +766,26 @@ private static function createDeadlineExpiredTask(string $runId): array is_string($run->cancellation_request_command_id) && $run->cancellation_deadline_at !== null && $now->gte($run->cancellation_deadline_at) + ) + || ( + ! is_string($run->cancellation_request_command_id) + && $run->timeout_cleanup_work_deadline_at !== null + && $now->gte($run->timeout_cleanup_work_deadline_at) ); if (! $deadlineExpired) { return null; } - if (is_string($run->cancellation_request_command_id) + // Enter under the run lock even when no queue worker is + // available, so hard-deadline enforcement needs no callback. + $enteredCleanup = TimeoutCleanup::enter($run, $now); + [, $hardDeadline] = TimeoutCleanup::earliestHardDeadline($run); + $timeoutCleanupExpired = $run->timeout_cleanup_work_deadline_at !== null + && $hardDeadline !== null && $now->gte($hardDeadline); + if ($timeoutCleanupExpired || (is_string($run->cancellation_request_command_id) && $run->cancellation_deadline_at !== null - && $now->gte($run->cancellation_deadline_at)) { + && $now->gte($run->cancellation_deadline_at))) { // Expiry is runtime authority. A foreign workflow definition // or an absent SDK worker must not keep the run open. $controlTask = $existingWorkflowTask ?? WorkflowTask::query()->create([ @@ -781,13 +802,22 @@ private static function createDeadlineExpiredTask(string $runId): array 'compatibility' => $run->compatibility, 'repair_count' => 1, ]); - $cancellationEnforced = app(WorkflowExecutor::class) - ->cancelIfCleanupDeadlineExpired($run, $controlTask); + $executor = app(WorkflowExecutor::class); + $cancellationEnforced = $timeoutCleanupExpired + ? $executor->timeoutIfDeadlineExpired($run, $controlTask) + : $executor->cancelIfCleanupDeadlineExpired($run, $controlTask); return null; } if ($existingWorkflowTask !== null) { + if ($enteredCleanup) { + self::historyProjectionRole()->projectRun( + $run->fresh( + ['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'] + ) + ); + } return null; } diff --git a/src/V2/WorkflowStub.php b/src/V2/WorkflowStub.php index fbd0cfc7a..31e912d85 100644 --- a/src/V2/WorkflowStub.php +++ b/src/V2/WorkflowStub.php @@ -60,6 +60,7 @@ use Workflow\V2\Support\CommandIntent; use Workflow\V2\Support\CommandRequestIdentity; use Workflow\V2\Support\ConfiguredV2Models; +use Workflow\V2\Support\CooperativeCancellationDelivery; use Workflow\V2\Support\CurrentRunResolver; use Workflow\V2\Support\ExecutionTracing; use Workflow\V2\Support\ExternalPayloads; @@ -87,6 +88,7 @@ use Workflow\V2\Support\TaskRepair; use Workflow\V2\Support\TaskSchedulingFields; use Workflow\V2\Support\TerminalCancellationHistory; +use Workflow\V2\Support\TimeoutCleanup; use Workflow\V2\Support\TimerCancellation; use Workflow\V2\Support\TraceCarrier; use Workflow\V2\Support\TypeRegistry; @@ -989,6 +991,7 @@ public function start(...$arguments): StartResult public function attemptStart(...$arguments): StartResult { [$arguments, $startOptions] = $this->extractStartArguments($arguments); + TimeoutCleanup::assertEmbeddedAdmission($startOptions->timeoutCleanupSeconds); $metadata = WorkflowMetadata::fromStartArguments($arguments); // Capture outside the storage transaction and preserve it across its retries. $observabilityContext = null; @@ -1055,6 +1058,7 @@ public function attemptSignalWithStart( $startArguments, DuplicateStartPolicy::ReturnExistingActive, ); + TimeoutCleanup::assertEmbeddedAdmission($startOptions->timeoutCleanupSeconds); if ($startOptions->duplicateStartPolicy !== DuplicateStartPolicy::ReturnExistingActive) { throw new LogicException( @@ -1806,6 +1810,15 @@ private function admitStart( 'run_timeout_seconds' => $runTimeoutSeconds, 'execution_deadline_at' => $executionDeadlineAt, 'run_deadline_at' => $runDeadlineAt, + // Only an opted-in run writes the column, so custom run tables without + // it keep working until they migrate (#863). + ...(($timeoutCleanupWorkDeadline = TimeoutCleanup::workDeadline( + $startOptions->timeoutCleanupSeconds, + $executionDeadlineAt, + $runDeadlineAt, + )) === null ? [] : [ + 'timeout_cleanup_work_deadline_at' => $timeoutCleanupWorkDeadline, + ]), 'status' => RunStatus::Pending->value, 'compatibility' => WorkerCompatibility::current(), 'payload_codec' => $payloadCodec, @@ -1847,6 +1860,11 @@ private function admitStart( 'run_timeout_seconds' => $runTimeoutSeconds, 'execution_deadline_at' => $executionDeadlineAt, 'run_deadline_at' => $runDeadlineAt, + TimeoutCleanup::STARTED_KEY => TimeoutCleanup::startedPayload( + $startOptions->timeoutCleanupSeconds, + $executionDeadlineAt, + $runDeadlineAt, + ), ]); $task = $this->createInitialWorkflowTask($run); @@ -2083,6 +2101,9 @@ private function recordWorkflowStartedHistory( 'run_timeout_seconds' => $startMetadata['run_timeout_seconds'], 'execution_deadline_at' => $startMetadata['execution_deadline_at']?->toIso8601String(), 'run_deadline_at' => $startMetadata['run_deadline_at']?->toIso8601String(), + ...(isset($startMetadata[TimeoutCleanup::STARTED_KEY]) ? [ + TimeoutCleanup::STARTED_KEY => $startMetadata[TimeoutCleanup::STARTED_KEY], + ] : []), 'workflow_definition_fingerprint' => WorkflowDefinition::fingerprint($workflowClass), 'declared_queries' => $commandContract['queries'], 'declared_query_contracts' => $commandContract['query_contracts'], @@ -2280,6 +2301,28 @@ private function recordCancellationRequest( if (is_string($run->cancellation_request_command_id)) { $existing = WorkflowCommand::query()->findOrFail($run->cancellation_request_command_id); $existingContext = (new CommandResult($existing))->cancellationContext(); + if ($existingContext?->isTimeout() === true && $parentContext === null && $this->runIsActive($run)) { + // Audit the user's request without replacing the engine's + // timeout interruption or renewing its admitted budget. + $codec = $run->payload_codec ?? CodecRegistry::defaultCodec(); + $command = WorkflowCommand::record($instance, $run, $this->commandAttributes([ + 'command_type' => CommandType::RequestCancellation->value, + 'target_scope' => $this->commandTargetScope(), + 'status' => CommandStatus::Accepted->value, + 'outcome' => CommandOutcome::CancellationRequested->value, + 'payload_codec' => $codec, + 'payload' => Serializer::serializeWithCodec($codec, [ + 'reason' => $reason, + 'existing_request_id' => $existing->id, + 'cleanup_deadline_at' => $run->cancellation_deadline_at?->toISOString(), + 'cancellation' => $existingContext->toArray(), + ]), + 'accepted_at' => now(), + 'applied_at' => now(), + ])); + + return; + } if ($parentContext !== null && $existingContext?->rootRequestId !== $parentContext->rootRequestId) { $codec = $run->payload_codec ?? CodecRegistry::defaultCodec(); $command = $this->rejectCommand( @@ -2366,6 +2409,12 @@ private function recordCancellationRequest( $requestedAt = now(); $deadline = $scopeDeadline ?? $scopeOrigin?->deadline() ?? $parentContext?->deadline() ?? $requestedAt->copy() ->addSeconds($cleanupTimeoutSeconds); + if ($run->timeout_cleanup_work_deadline_at !== null) { + [, $hardDeadline] = TimeoutCleanup::earliestHardDeadline($run); + if ($hardDeadline !== null && $hardDeadline->lt($deadline)) { + $deadline = $hardDeadline; + } + } $reason = $parentContext?->reason ?? $reason; $requestId = (string) Str::ulid(); $caller = $this->resolvedCommandContext() @@ -3636,6 +3685,15 @@ private function admitSignalWithNewRun( 'run_timeout_seconds' => $runTimeoutSeconds, 'execution_deadline_at' => $executionDeadlineAt, 'run_deadline_at' => $runDeadlineAt, + // Only an opted-in run writes the column, so custom run tables without + // it keep working until they migrate (#863). + ...(($timeoutCleanupWorkDeadline = TimeoutCleanup::workDeadline( + $startOptions->timeoutCleanupSeconds, + $executionDeadlineAt, + $runDeadlineAt, + )) === null ? [] : [ + 'timeout_cleanup_work_deadline_at' => $timeoutCleanupWorkDeadline, + ]), 'status' => RunStatus::Pending->value, 'compatibility' => WorkerCompatibility::current(), 'payload_codec' => $payloadCodec, @@ -3683,6 +3741,11 @@ private function admitSignalWithNewRun( 'run_timeout_seconds' => $runTimeoutSeconds, 'execution_deadline_at' => $executionDeadlineAt, 'run_deadline_at' => $runDeadlineAt, + TimeoutCleanup::STARTED_KEY => TimeoutCleanup::startedPayload( + $startOptions->timeoutCleanupSeconds, + $executionDeadlineAt, + $runDeadlineAt, + ), ]); $task = $this->createInitialWorkflowTask($run); @@ -4186,6 +4249,12 @@ private function attemptTerminalCommand( $terminalHistoryPayload['reason'] = $reason; } + // A hard cancel or terminate ends deadline-triggered cleanup early (#863). + if (is_string($run->cancellation_request_command_id) + && CooperativeCancellationDelivery::context($run)?->isTimeout() === true) { + $terminalHistoryPayload[TimeoutCleanup::STARTED_KEY] = TimeoutCleanup::outcome($run, 'preempted'); + } + WorkflowHistoryEvent::record($run, $terminalEventType, $terminalHistoryPayload, null, $command); PendingUpdateCloser::closeForTerminalRun($run); diff --git a/src/migrations/2026_11_13_000100_add_timeout_cleanup_work_deadline_to_workflow_runs.php b/src/migrations/2026_11_13_000100_add_timeout_cleanup_work_deadline_to_workflow_runs.php new file mode 100644 index 000000000..62178860d --- /dev/null +++ b/src/migrations/2026_11_13_000100_add_timeout_cleanup_work_deadline_to_workflow_runs.php @@ -0,0 +1,45 @@ +withSchemaLockBudget(static function (): void { + Schema::table('workflow_runs', static function (Blueprint $table): void { + $table->timestamp('timeout_cleanup_work_deadline_at', 6) + ->nullable(); + }); + }); + } + + $this->createIndexOnline('workflow_runs', 'workflow_runs_timeout_cleanup_work_deadline_idx', [ + 'timeout_cleanup_work_deadline_at', + ]); + } + + public function down(): void + { + Schema::table('workflow_runs', static function (Blueprint $table): void { + $table->dropIndex('workflow_runs_timeout_cleanup_work_deadline_idx'); + $table->dropColumn('timeout_cleanup_work_deadline_at'); + }); + } +}; diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index f4f65839f..615b3faaf 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -261,6 +261,172 @@ public function testRedriveStartVisibilityCorpusCarriesToTheSuccessor(): void ); } + public function testTimeoutCleanupOutcomeCorpusEndsTimedOutWithEachOutcome(): void + { + $path = self::FIXTURE_DIR . '/timeout-cleanup-outcome-recorded.json'; + if (! is_file($path)) { + $this->markTestSkipped('The timeout cleanup outcome evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['timeout_cleanup_outcome']; + config([ + 'queue.default' => 'database', + ]); + Queue::fake(); + + foreach ($scenario['outcomes'] as $mode => $outcome) { + $this->travelTo(now()->startOfSecond()); + $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-timeout-cleanup-' . $mode); + $stub->start($mode, new StartOptions( + executionTimeoutSeconds: $scenario['execution_timeout_seconds'], + timeoutCleanupSeconds: $scenario['timeout_cleanup_seconds'], + )); + $runId = $stub->runId(); + $this->runReadyTaskOfType($runId, TaskType::Workflow); + $workSeconds = $scenario['execution_timeout_seconds'] - $scenario['timeout_cleanup_seconds']; + $this->travel($workSeconds + 1) + ->seconds(); + \Workflow\V2\TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyTaskOfType($runId, TaskType::Workflow); + if ($outcome === 'expired') { + $this->travel($scenario['timeout_cleanup_seconds'])->seconds(); + \Workflow\V2\TaskWatchdog::runPass(runIds: [$runId]); + } + + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowTimedOut, $terminal->event_type, $mode); + $this->assertSame($outcome, $terminal->payload['timeout_cleanup']['outcome'] ?? null, $mode); + $this->assertTrue(WorkflowHistoryEvent::query() + ->where('workflow_run_id', $runId) + ->where('event_type', HistoryEventType::WorkflowTimeoutCleanupStarted->value) + ->exists(), $mode); + } + } + + public function testTimeoutCleanupTaskExpiryCorpusKeepsTheOutcome(): void + { + $path = self::FIXTURE_DIR . '/timeout-cleanup-task-expiry.json'; + if (! is_file($path)) { + $this->markTestSkipped('The timeout cleanup task expiry evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['timeout_cleanup_task_expiry']; + config() + ->set('queue.default', 'database'); + Queue::fake(); + $this->travelTo(now()->startOfSecond()); + $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-timeout-cleanup-task-expiry'); + $stub->start($fixture['workflow']['arguments'][0], new StartOptions( + executionTimeoutSeconds: $scenario['execution_timeout_seconds'], + timeoutCleanupSeconds: $scenario['timeout_cleanup_seconds'], + )); + $runId = $stub->runId(); + $this->runReadyTaskOfType($runId, TaskType::Workflow); + $this->travel($scenario['detect_after_seconds'])->seconds(); + \Workflow\V2\TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyTaskOfType($runId, TaskType::Workflow); + $run = WorkflowRun::query()->findOrFail($runId); + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $runId, + 'namespace' => $run->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Ready, + 'available_at' => now(), + 'connection' => $run->connection, + 'queue' => $run->queue, + 'compatibility' => $run->compatibility, + 'payload' => [ + 'reason' => 'cleanup_resume', + ], + ]); + $this->travel($scenario['execution_timeout_seconds'] - $scenario['detect_after_seconds'])->seconds(); + // The embedded task boundary must agree with watchdog expiry. + $this->app->call([new RunWorkflowTask($task->id), 'handle']); + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowTimedOut, $terminal->event_type); + $this->assertSame($scenario['outcome'], $terminal->payload['timeout_cleanup']['outcome'] ?? null); + $this->assertSame( + $run->execution_deadline_at->toIso8601String(), + $terminal->payload['timeout_cleanup']['cleanup_deadline_at'] ?? null + ); + } + + public function testTimeoutCleanupContinueAsNewCorpusKeepsTheWorkDeadline(): void + { + $path = self::FIXTURE_DIR . '/timeout-cleanup-continue-as-new-inheritance.json'; + if (! is_file($path)) { + $this->markTestSkipped('The timeout cleanup continue-as-new evidence is absent.'); + } + config([ + 'queue.default' => 'database', + ]); + Queue::fake(); + $this->travelTo(now()->startOfSecond()); + + $stub = WorkflowStub::make( + \Tests\Fixtures\V2\TestContinueAsNewLargePayloadWorkflow::class, + 'regression-corpus-timeout-cleanup-continue', + ); + $stub->start('x', 0, new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $first = WorkflowRun::query()->findOrFail($stub->runId()); + $this->runReadyTaskOfType($first->id, TaskType::Workflow); + $second = WorkflowRun::query() + ->where('workflow_instance_id', $first->workflow_instance_id) + ->where('run_number', 2) + ->sole(); + + $this->assertNotNull($first->timeout_cleanup_work_deadline_at); + $this->assertTrue($second->timeout_cleanup_work_deadline_at->equalTo($first->timeout_cleanup_work_deadline_at)); + $this->assertSame(10, WorkflowHistoryEvent::query() + ->where('workflow_run_id', $second->id) + ->where('event_type', HistoryEventType::WorkflowStarted->value) + ->sole() + ->payload['timeout_cleanup']['allowance_seconds'] ?? null); + } + + public function testTimeoutCleanupRedriveCorpusKeepsTheWorkDeadline(): void + { + $path = self::FIXTURE_DIR . '/timeout-cleanup-redrive-inheritance.json'; + if (! is_file($path)) { + $this->markTestSkipped('The timeout cleanup redrive evidence is absent.'); + } + self::stopWorkers(); + config([ + 'queue.default' => 'sync', + 'queue.connections.sync.driver' => 'sync', + ]); + Cache::forget('test:redrive:first-calls'); + Cache::forget('test:redrive:second-calls'); + $this->clearWorkflowState(); + WorkflowStub::fake(); + + $stub = WorkflowStub::make( + \Tests\Fixtures\V2\TestRedriveWorkflow::class, + 'regression-corpus-timeout-cleanup-redrive' + ); + $stub->start('Taylor', new StartOptions(executionTimeoutSeconds: 3600, timeoutCleanupSeconds: 60)); + $this->assertTrue($stub->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($stub->runId()); + + $result = app(WorkflowControlPlane::class)->redrive($source->workflow_instance_id, $source->id); + + $this->assertTrue($result['accepted'], (string) $result['reason']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $this->assertNotNull($source->timeout_cleanup_work_deadline_at); + $this->assertTrue( + $successor->timeout_cleanup_work_deadline_at->equalTo($source->timeout_cleanup_work_deadline_at) + ); + $this->assertSame(60, WorkflowHistoryEvent::query() + ->where('workflow_run_id', $successor->id) + ->where('event_type', HistoryEventType::WorkflowStarted->value) + ->sole() + ->payload['timeout_cleanup']['allowance_seconds'] ?? null); + } + public function testRedriveServiceFenceCorpusRefusesUnqualifiedReuse(): void { $path = self::FIXTURE_DIR . '/redrive-service-unqualified-history.json'; diff --git a/tests/Feature/V2/V2TimeoutCleanupAdmissionTest.php b/tests/Feature/V2/V2TimeoutCleanupAdmissionTest.php new file mode 100644 index 000000000..70c31c680 --- /dev/null +++ b/tests/Feature/V2/V2TimeoutCleanupAdmissionTest.php @@ -0,0 +1,159 @@ +set('queue.default', 'redis'); + } + + public function testStartRecordsTheAllowanceAndTheDerivedWorkDeadline(): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestContinueAsNewLargePayloadWorkflow::class, 'timeout-cleanup-admission'); + $workflow->start('x', 1, new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $run = WorkflowRun::query()->findOrFail($workflow->runId()); + + $this->assertTrue( + $run->timeout_cleanup_work_deadline_at->equalTo($run->execution_deadline_at->copy()->subSeconds(10)) + ); + $started = $this->started($run->id); + $this->assertSame(10, $started->payload['timeout_cleanup']['allowance_seconds']); + $this->assertSame( + $run->timeout_cleanup_work_deadline_at->toIso8601String(), + $started->payload['timeout_cleanup']['work_deadlines']['execution'], + ); + } + + public function testAStartWithoutTheOptionRecordsNothing(): void + { + $workflow = WorkflowStub::make(TestContinueAsNewLargePayloadWorkflow::class, 'timeout-cleanup-none'); + $workflow->start('x', 1, new StartOptions(executionTimeoutSeconds: 70)); + $run = WorkflowRun::query()->findOrFail($workflow->runId()); + + $this->assertNull($run->timeout_cleanup_work_deadline_at); + $this->assertArrayNotHasKey('timeout_cleanup', $this->started($run->id)->payload); + } + + public function testContinueAsNewKeepsTheExecutionWorkDeadline(): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestContinueAsNewLargePayloadWorkflow::class, 'timeout-cleanup-continue'); + $workflow->start('x', 0, new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $first = WorkflowRun::query()->findOrFail($workflow->runId()); + + $this->travel(5) + ->seconds(); + $task = WorkflowTask::query() + ->where('workflow_run_id', $first->id) + ->where('task_type', TaskType::Workflow->value) + ->where('status', TaskStatus::Ready->value) + ->sole(); + $this->app->call([new RunWorkflowTask($task->id), 'handle']); + + $second = WorkflowRun::query() + ->where('workflow_instance_id', $first->workflow_instance_id) + ->where('run_number', 2) + ->sole(); + $this->assertTrue($second->execution_deadline_at->equalTo($first->execution_deadline_at)); + $this->assertTrue($second->timeout_cleanup_work_deadline_at->equalTo($first->timeout_cleanup_work_deadline_at)); + $this->assertSame(10, $this->started($second->id)->payload['timeout_cleanup']['allowance_seconds']); + } + + public function testContinueAsNewDerivesANewRunWorkDeadline(): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestContinueAsNewLargePayloadWorkflow::class, 'timeout-cleanup-continue-run'); + $workflow->start('x', 0, new StartOptions(runTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $first = WorkflowRun::query()->findOrFail($workflow->runId()); + $this->travel(30) + ->seconds(); + $task = WorkflowTask::query()->where('workflow_run_id', $first->id) + ->where('task_type', TaskType::Workflow->value)->where('status', TaskStatus::Ready->value)->sole(); + $this->app->call([new RunWorkflowTask($task->id), 'handle']); + $second = WorkflowRun::query()->where('workflow_instance_id', $first->workflow_instance_id) + ->where('run_number', 2) + ->sole(); + $this->assertNull($second->execution_deadline_at); + $this->assertTrue($second->run_deadline_at->equalTo(now()->addSeconds(70))); + $this->assertTrue($second->timeout_cleanup_work_deadline_at->equalTo(now()->addSeconds(60))); + $this->assertSame( + $second->timeout_cleanup_work_deadline_at->toIso8601String(), + $this->started($second->id) +->payload['timeout_cleanup']['work_deadlines']['run'] + ); + } + + public function testPollModeRefusesTimeoutCleanupBeforeStartWrites(): void + { + $workflow = WorkflowStub::make(TestContinueAsNewLargePayloadWorkflow::class, 'timeout-cleanup-poll'); + config() + ->set('workflows.v2.task_dispatch_mode', 'poll'); + $this->expectException(\LogicException::class); + $this->expectExceptionMessage('timeout cleanup is not supported by poll workers'); + try { + $workflow->start('x', 1, new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + } finally { + $this->assertSame(0, WorkflowRun::query()->count()); + $this->assertSame(0, \Workflow\V2\Models\WorkflowCommand::query()->count()); + } + } + + public function testTheBridgeRefusesAnEmbeddedTimeoutCleanupRunWithoutTakingItsLease(): void + { + $workflow = WorkflowStub::make(TestContinueAsNewLargePayloadWorkflow::class, 'timeout-cleanup-bridge'); + $workflow->start('x', 1, new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $task = WorkflowTask::query()->where('workflow_run_id', $workflow->runId())->sole(); + $bridge = $this->app->make(\Workflow\V2\Contracts\WorkflowTaskBridge::class); + $result = $bridge->claimStatus($task->id, 'unsupported-worker'); + $this->assertFalse($result['claimed']); + $this->assertSame('timeout_cleanup_not_supported', $result['reason']); + $this->assertSame(TaskStatus::Ready, $task->refresh()->status); + $this->assertNull($task->lease_owner); + } + + public function testTheServiceStartRefusesTheOptionBeforeCreatingAnInstance(): void + { + $control = $this->app->make(\Workflow\V2\Contracts\WorkflowControlPlane::class); + $result = $control->start('foreign-workflow', 'timeout-cleanup-service', [ + 'execution_timeout_seconds' => 70, + 'timeout_cleanup_seconds' => 10, + ]); + $this->assertFalse($result['accepted']); + $this->assertSame('timeout_cleanup_not_supported', $result['reason']); + $this->assertSame(0, \Workflow\V2\Models\WorkflowInstance::query()->count()); + } + + private function started(string $runId): WorkflowHistoryEvent + { + return WorkflowHistoryEvent::query() + ->where('workflow_run_id', $runId) + ->where('event_type', HistoryEventType::WorkflowStarted->value) + ->sole(); + } +} diff --git a/tests/Feature/V2/V2TimeoutCleanupMigrationTest.php b/tests/Feature/V2/V2TimeoutCleanupMigrationTest.php new file mode 100644 index 000000000..1a94fcb0e --- /dev/null +++ b/tests/Feature/V2/V2TimeoutCleanupMigrationTest.php @@ -0,0 +1,76 @@ +migration(); + $this->assertFalse($migration->withinTransaction); + + $source = (string) file_get_contents(self::path()); + $this->assertStringContainsString('withSchemaLockBudget', $source); + $this->assertStringContainsString('createIndexOnline', $source); + // No backfill: existing runs never opted in, so nothing is rewritten. + $this->assertDoesNotMatchRegularExpression('/->\s*(update|insert|upsert|delete)\s*\(/', $source); + } + + public function testItAddsANullableIndexedColumnToAPopulatedTableAndCanRunAgain(): void + { + $migration = $this->migration(); + $migration->down(); + $this->assertFalse(Schema::hasColumn('workflow_runs', 'timeout_cleanup_work_deadline_at')); + + $ids = []; + foreach (range(1, 5) as $number) { + $ids[] = $id = (string) Str::ulid(); + DB::table('workflow_runs')->insert([ + 'id' => $id, + 'workflow_instance_id' => 'migration-instance-' . $number, + 'run_number' => 1, + 'workflow_class' => 'App\\Workflows\\Example', + 'workflow_type' => 'example', + 'status' => 'running', + 'created_at' => now(), + 'updated_at' => now(), + ]); + } + + $migration->up(); + // An interrupted or repeated run skips work already done. + $migration->up(); + + $this->assertTrue(Schema::hasColumn('workflow_runs', 'timeout_cleanup_work_deadline_at')); + $this->assertTrue(Schema::hasIndex('workflow_runs', 'workflow_runs_timeout_cleanup_work_deadline_idx')); + $this->assertSame( + 5, + DB::table('workflow_runs')->whereIn('id', $ids)->whereNull('timeout_cleanup_work_deadline_at')->count(), + ); + } + + private function migration(): WorkflowMigration + { + return require self::path(); + } + + private static function path(): string + { + return dirname(__DIR__, 3) . '/src/migrations/' . self::NAME . '.php'; + } +} diff --git a/tests/Feature/V2/V2TimeoutCleanupTest.php b/tests/Feature/V2/V2TimeoutCleanupTest.php new file mode 100644 index 000000000..15fef9d85 --- /dev/null +++ b/tests/Feature/V2/V2TimeoutCleanupTest.php @@ -0,0 +1,536 @@ +set('queue.default', 'redis'); + } + + /** + * @return iterable + */ + public static function outcomes(): iterable + { + yield 'cleanup completes' => ['complete', 0, 'completed']; + yield 'cleanup rethrows the interruption' => ['rethrow', 0, 'completed']; + yield 'cleanup throws' => ['fail', 0, 'failed']; + yield 'cleanup outlives the hard deadline' => ['expire', 11, 'expired']; + } + + #[DataProvider('outcomes')] + public function testAnIdleRunIsWokenAtItsWorkDeadlineAndEndsTimedOut( + string $mode, + int $extraSeconds, + string $outcome, + ): void { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestTimeoutCleanupWorkflow::class, 'timeout-cleanup-' . $mode); + $workflow->start($mode, new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $runId = $workflow->runId(); + $this->runReadyWorkflowTask($runId); + $this->assertSame(RunStatus::Waiting, WorkflowRun::query()->findOrFail($runId)->status); + + // Past the work deadline, before the hard deadline: the watchdog wakes it. + $this->travel(61) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyWorkflowTask($runId); + + if ($extraSeconds > 0) { + $this->assertFalse(WorkflowRun::query()->findOrFail($runId)->status->isTerminal()); + $this->travel($extraSeconds) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyWorkflowTaskIfAny($runId); + } + + $run = WorkflowRun::query()->findOrFail($runId); + $this->assertSame(RunStatus::Failed, $run->status); + $this->assertSame('timed_out', $run->closed_reason); + + $events = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->orderBy('sequence')->get(); + $started = $events->firstWhere('event_type', HistoryEventType::WorkflowTimeoutCleanupStarted); + $this->assertNotNull($started); + $this->assertSame('execution', $started->payload['timeout_kind']); + $terminal = $events->last(); + $this->assertSame(HistoryEventType::WorkflowTimedOut, $terminal->event_type); + $this->assertSame($outcome, $terminal->payload['timeout_cleanup']['outcome']); + $this->assertSame('execution_timeout', $terminal->payload['timeout_kind']); + $this->assertSame('timeout', $terminal->payload['failure_category']); + if ($outcome === 'failed') { + $this->assertIsString($terminal->payload['timeout_cleanup']['failure_id'] ?? null); + } + // Describe, the export and the failure list read this message. + $expected = [ + 'completed' => 'Timeout cleanup completed.', + 'failed' => 'Timeout cleanup failed.', + 'expired' => 'Timeout cleanup did not finish before the hard deadline.', + ][$outcome]; + $this->assertStringEndsWith($expected, $terminal->payload['message']); + $this->assertStringEndsWith($expected, \Workflow\V2\Models\WorkflowFailure::query() + ->findOrFail($terminal->payload['failure_id'])->message); + } + + public function testTerminatingDuringCleanupPreemptsIt(): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestTimeoutCleanupWorkflow::class, 'timeout-cleanup-preempted'); + $workflow->start('expire', new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $runId = $workflow->runId(); + $this->runReadyWorkflowTask($runId); + $this->travel(61) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyWorkflowTask($runId); + $this->assertFalse(WorkflowRun::query()->findOrFail($runId)->status->isTerminal()); + + $workflow->refresh() + ->terminate('operator stop'); + + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowTerminated, $terminal->event_type); + $this->assertSame('preempted', $terminal->payload['timeout_cleanup']['outcome'] ?? null); + $this->assertSame(RunStatus::Terminated, WorkflowRun::query()->findOrFail($runId)->status); + } + + public function testWithoutTheOptionTheHardTimeoutIsUnchanged(): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestTimeoutCleanupWorkflow::class, 'timeout-cleanup-off'); + $workflow->start('complete', new StartOptions(executionTimeoutSeconds: 70)); + $runId = $workflow->runId(); + $this->runReadyWorkflowTask($runId); + + $this->travel(61) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->assertSame(0, WorkflowTask::query() + ->where('workflow_run_id', $runId) + ->where('status', TaskStatus::Ready->value) + ->where('task_type', TaskType::Workflow->value) + ->count()); + + $this->travel(10) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyWorkflowTaskIfAny($runId); + + $events = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->orderBy('sequence')->get(); + $this->assertNull($events->firstWhere('event_type', HistoryEventType::WorkflowTimeoutCleanupStarted)); + $this->assertSame(HistoryEventType::WorkflowTimedOut, $events->last()->event_type); + $this->assertArrayNotHasKey('timeout_cleanup', $events->last()->payload); + } + + public function testTaskCompletionAtTheHardDeadlineRecordsExpiredCleanup(): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestTimeoutCleanupWorkflow::class, 'timeout-cleanup-task-expired'); + $workflow->start('expire', new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $runId = $workflow->runId(); + $this->runReadyWorkflowTask($runId); + $this->travel(61) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyWorkflowTask($runId); + + $run = WorkflowRun::query()->findOrFail($runId); + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $runId, + 'namespace' => $run->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Ready, + 'available_at' => now(), + 'connection' => $run->connection, + 'queue' => $run->queue, + 'compatibility' => $run->compatibility, + 'payload' => [ + 'reason' => 'cleanup_resume', + ], + ]); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $task->forceFill([ + 'status' => TaskStatus::Leased, + 'leased_at' => now(), + 'lease_expires_at' => now() + ->addSeconds(60), + 'lease_owner' => 'old-worker', + ])->save(); + $this->travel(9) + ->seconds(); + $result = $bridge->complete($task->id, [[ + 'type' => 'complete_workflow', + 'result' => 'late', + ]]); + $this->assertFalse($result['completed']); + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowTimedOut, $terminal->event_type); + $this->assertSame('expired', $terminal->payload['timeout_cleanup']['outcome'] ?? null); + } + + /** + * @return iterable + */ + public static function replayErrors(): iterable + { + yield 'version range' => ['version-mismatch']; + yield 'history shape' => ['history-mismatch']; + yield 'unresolved failure' => ['unresolved-failure']; + } + + #[DataProvider('replayErrors')] + public function testReplayErrorsDuringCleanupRemainRepairable(string $mode): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestTimeoutCleanupWorkflow::class, 'timeout-cleanup-' . $mode); + $workflow->start($mode, new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $runId = $workflow->runId(); + $this->runReadyWorkflowTask($runId); + $this->travel(61) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyWorkflowTask($runId); + + $this->assertFalse(WorkflowRun::query()->findOrFail($runId)->status->isTerminal()); + $this->assertSame(0, WorkflowHistoryEvent::query()->where('workflow_run_id', $runId) + ->where('event_type', HistoryEventType::WorkflowTimedOut->value)->count()); + } + + public function testLateDetectionAndReplacementDoNotRenewOrRestartCleanup(): void + { + [$workflow, $run] = $this->enterCleanupAt(67); + $started = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::WorkflowTimeoutCleanupStarted->value)->sole(); + $this->assertSame(now()->toIso8601String(), $started->payload['detected_at']); + $this->assertSame($run->execution_deadline_at->toIso8601String(), $started->payload['cleanup_deadline_at']); + $deadline = $run->cancellation_deadline_at; + + $this->newResumeTask($run); + $this->runReadyWorkflowTask($run->id); + TaskWatchdog::runPass(runIds: [$run->id]); + $this->assertSame(1, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::WorkflowTimeoutCleanupStarted->value)->count()); + $this->assertTrue($run->refresh()->cancellation_deadline_at->equalTo($deadline)); + } + + public function testTheEarliestHardDeadlineOwnsCleanup(): void + { + [$workflow, $run] = $this->enterCleanupAt(30, new StartOptions( + executionTimeoutSeconds: 70, + runTimeoutSeconds: 40, + timeoutCleanupSeconds: 10, + )); + $started = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::WorkflowTimeoutCleanupStarted->value)->sole(); + $this->assertSame('run', $started->payload['timeout_kind']); + $this->assertSame($run->run_deadline_at->toIso8601String(), $started->payload['cleanup_deadline_at']); + } + + public function testUserCancellationIsAuditedWithoutReplacingTheTimeoutRequest(): void + { + [$workflow, $run] = $this->enterCleanupAt(61); + $requestId = $run->cancellation_request_command_id; + $deadline = $run->cancellation_deadline_at; + $result = $workflow->refresh() + ->attemptRequestCancellation('operator request', 120); + $this->assertTrue($result->accepted()); + $this->assertNotSame($requestId, $result->commandId()); + $this->assertSame($requestId, $result->payloadValues(['existing_request_id'])['existing_request_id']); + $this->assertSame('operator request', $result->reason()); + $this->assertSame($requestId, $run->refresh()->cancellation_request_command_id); + $this->assertTrue($run->cancellation_deadline_at->equalTo($deadline)); + } + + public function testHardCancelPreemptsCleanup(): void + { + [$workflow, $run] = $this->enterCleanupAt(61); + $workflow->refresh() + ->cancel('operator stop'); + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowCancelled, $terminal->event_type); + $this->assertSame('preempted', $terminal->payload['timeout_cleanup']['outcome'] ?? null); + } + + public function testEarlierCooperativeCancellationIsPreemptedWhenTheHardDeadlineWins(): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestTimeoutCleanupWorkflow::class, 'timeout-cleanup-earlier-cancel'); + $workflow->start('expire', new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $runId = $workflow->runId(); + $this->runReadyWorkflowTask($runId); + $this->travel(20) + ->seconds(); + $workflow->refresh() + ->requestCancellation('earlier request', 120); + $run = WorkflowRun::query()->findOrFail($runId); + $this->assertTrue($run->cancellation_deadline_at->equalTo($run->execution_deadline_at)); + $this->travel(50) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowTimedOut, $terminal->event_type); + $this->assertSame('preempted', $terminal->payload['cancellation_cleanup']['outcome'] ?? null); + $this->assertFalse(WorkflowHistoryEvent::query()->where('workflow_run_id', $runId) + ->where('event_type', HistoryEventType::WorkflowTimeoutCleanupStarted->value)->exists()); + } + + public function testTimeoutCleanupListsAnAbandonedActivityAndAcceptsNoLateHistory(): void + { + [$workflow, $run] = $this->enterCleanupAt(61); + $task = $this->newResumeTask($run); + // Internal abandon lifetime, as in V2PortableCancellationDeliveryTest. + // Public embedded authoring does not expose this policy yet. + $execution = \Workflow\V2\Models\ActivityExecution::query()->create([ + 'workflow_run_id' => $run->id, + 'sequence' => 3, + 'activity_class' => \Tests\Fixtures\V2\TestGreetingActivity::class, + 'activity_type' => \Tests\Fixtures\V2\TestGreetingActivity::class, + 'status' => \Workflow\V2\Enums\ActivityStatus::Pending, + 'arguments' => \Workflow\Serializers\Serializer::serialize(['cleanup']), + 'connection' => $run->connection, + 'queue' => $run->queue, + 'activity_options' => [ + 'cancellation_policy' => 'abandon', + ], + 'retry_policy' => [ + 'snapshot_version' => 1, + 'max_attempts' => 1, + 'backoff_seconds' => [0], + 'schedule_to_close_timeout' => 180, + ], + 'schedule_to_close_deadline_at' => now() + ->addSeconds(180), + ]); + WorkflowHistoryEvent::record($run, HistoryEventType::ActivityScheduled, [ + 'activity_execution_id' => $execution->id, + 'activity_class' => $execution->activity_class, + 'activity_type' => $execution->activity_type, + 'sequence' => 3, + 'activity' => \Workflow\V2\Support\ActivitySnapshot::fromExecution($execution), + ], $task); + $activityTask = WorkflowTask::query()->create([ + 'workflow_run_id' => $run->id, + 'namespace' => $run->namespace, + 'task_type' => TaskType::Activity, + 'status' => TaskStatus::Ready, + 'available_at' => now(), + 'connection' => $run->connection, + 'queue' => $run->queue, + 'compatibility' => $run->compatibility, + 'payload' => [ + 'activity_execution_id' => $execution->id, + ], + ]); + $this->travel(9) + ->seconds(); + TaskWatchdog::runPass(runIds: [$run->id]); + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowTimedOut, $terminal->event_type); + $this->assertSame('expired', $terminal->payload['timeout_cleanup']['outcome']); + $this->assertEquals([[ + 'activity_execution_id' => $execution->id, + 'sequence' => 3, + ]], $terminal->payload['abandoned_activities']); + $this->assertSame(\Workflow\V2\Enums\ActivityStatus::Pending, $execution->refresh()->status); + $count = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count(); + $activities = $this->app->make(\Workflow\V2\Contracts\ActivityTaskBridge::class); + $claim = $activities->claimStatus($activityTask->id, 'independent-cleanup-worker'); + $this->assertTrue($claim['claimed']); + $this->assertTrue($activities->heartbeat($claim['activity_attempt_id'])['can_continue']); + $outcome = $activities->complete( + $claim['activity_attempt_id'], + \Workflow\Serializers\Serializer::serialize('done') + ); + $this->assertFalse($outcome['recorded']); + $this->assertSame('run_closed', $outcome['reason']); + $this->assertSame($count, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + } + + public function testAScopedChildOfATimeoutRootCancelsWithoutItsOwnTimeout(): void + { + [$parent, $parentRun] = $this->enterCleanupAt(61); + $root = \Workflow\V2\Support\CooperativeCancellationDelivery::context($parentRun); + $child = WorkflowStub::make(\Tests\Fixtures\V2\TestTimerWorkflow::class, 'timeout-cleanup-scoped-child'); + $child->start(3600); + $childRun = WorkflowRun::query()->findOrFail($child->runId()); + $this->runReadyWorkflowTask($childRun->id); + $requestId = (string) \Illuminate\Support\Str::ulid(); + $context = \Workflow\V2\CancellationContext::fromScopeContext( + \Workflow\V2\ScopedCancellationContext::fromRunContext($root), + $requestId, + $childRun->workflow_instance_id, + $childRun->id, + ); + $command = \Workflow\V2\Models\WorkflowCommand::record($childRun->instance, $childRun, [ + 'id' => $requestId, + 'command_type' => \Workflow\V2\Enums\CommandType::RequestCancellation, + 'target_scope' => 'run', + 'status' => \Workflow\V2\Enums\CommandStatus::Accepted, + 'outcome' => \Workflow\V2\Enums\CommandOutcome::CancellationRequested, + 'payload_codec' => 'avro', + 'payload' => \Workflow\Serializers\Serializer::serialize([ + 'cancellation' => $context->toArray(), + ]), + 'accepted_at' => now(), + 'applied_at' => now(), + ]); + WorkflowHistoryEvent::record($childRun, HistoryEventType::CooperativeCancellationRequested, [ + 'workflow_command_id' => $command->id, + 'workflow_instance_id' => $childRun->workflow_instance_id, + 'workflow_run_id' => $childRun->id, + 'command_type' => 'request_cancellation', + 'reason' => $context->reason, + 'cleanup_deadline_at' => $context->deadline() + ->toISOString(), + 'cancellation' => $context->toArray(), + ], null, $command); + $childRun->forceFill([ + 'cancellation_request_command_id' => $command->id, + 'cancellation_requested_at' => now(), + 'cancellation_deadline_at' => $context->deadline(), + ])->save(); + $this->newResumeTask($childRun); + $this->runReadyWorkflowTask($childRun->id); + $this->assertNull($childRun->refresh()->execution_deadline_at); + $this->assertNull($childRun->run_deadline_at); + $this->assertSame(RunStatus::Cancelled, $childRun->status); + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $childRun->id)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowCancelled, $terminal->event_type); + $this->assertArrayNotHasKey('timeout_cleanup', $terminal->payload); + } + + /** + * @return iterable + */ + public static function unavailableWorkers(): iterable + { + yield 'detected during the cleanup window' => [false]; + yield 'first detected after the hard deadline' => [true]; + } + + #[DataProvider('unavailableWorkers')] + public function testUnavailableWorkersCannotKeepAnOptedInRunOpen(bool $lateDetection): void + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestTimeoutCleanupWorkflow::class, 'timeout-cleanup-no-worker'); + $workflow->start('expire', new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10)); + $runId = $workflow->runId(); + // Leave even the initial task Ready: no worker ever executes it. + $this->travel($lateDetection ? 71 : 61) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + if (! $lateDetection) { + $this->assertFalse(WorkflowRun::query()->findOrFail($runId)->status->isTerminal()); + $this->travel(9) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + } + $this->assertSame(RunStatus::Failed, WorkflowRun::query()->findOrFail($runId)->status); + $terminal = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->orderByDesc( + 'sequence' + )->firstOrFail(); + $this->assertSame(HistoryEventType::WorkflowTimedOut, $terminal->event_type); + $this->assertSame('expired', $terminal->payload['timeout_cleanup']['outcome'] ?? null); + $count = WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->count(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->assertSame($count, WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)->count()); + } + + /** + * @return array{WorkflowStub, WorkflowRun} + */ + private function enterCleanupAt(int $seconds, ?StartOptions $options = null): array + { + $this->travelTo(now()->startOfSecond()); + $workflow = WorkflowStub::make(TestTimeoutCleanupWorkflow::class, 'timeout-cleanup-boundary'); + $workflow->start( + 'expire', + $options ?? new StartOptions(executionTimeoutSeconds: 70, timeoutCleanupSeconds: 10) + ); + $runId = $workflow->runId(); + $this->runReadyWorkflowTask($runId); + $this->travel($seconds) + ->seconds(); + TaskWatchdog::runPass(runIds: [$runId]); + $this->runReadyWorkflowTask($runId); + return [$workflow, WorkflowRun::query()->findOrFail($runId)]; + } + + private function newResumeTask(WorkflowRun $run): WorkflowTask + { + return WorkflowTask::query()->create([ + 'workflow_run_id' => $run->id, + 'namespace' => $run->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Ready, + 'available_at' => now(), + 'connection' => $run->connection, + 'queue' => $run->queue, + 'compatibility' => $run->compatibility, + 'payload' => [ + 'reason' => 'cleanup_resume', + ], + ]); + } + + private function runReadyWorkflowTask(string $runId): void + { + $task = WorkflowTask::query() + ->where('workflow_run_id', $runId) + ->where('task_type', TaskType::Workflow->value) + ->where('status', TaskStatus::Ready->value) + ->sole(); + $this->app->call([new RunWorkflowTask($task->id), 'handle']); + } + + private function runReadyWorkflowTaskIfAny(string $runId): void + { + $task = WorkflowTask::query() + ->where('workflow_run_id', $runId) + ->where('task_type', TaskType::Workflow->value) + ->where('status', TaskStatus::Ready->value) + ->first(); + if ($task !== null) { + $this->app->call([new RunWorkflowTask($task->id), 'handle']); + } + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-continue-as-new-inheritance.json b/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-continue-as-new-inheritance.json new file mode 100644 index 000000000..1d1ff7edd --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-continue-as-new-inheritance.json @@ -0,0 +1,50 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "timeout-cleanup-continue-as-new-inheritance", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestTimeoutCleanupWorkflow", + "arguments": [ + "continue" + ], + "payload_codec": "avro" + }, + "timeout_cleanup_inheritance": { + "issue": "durable-workflow/workflow#863", + "path": "continue_as_new", + "note": "A continued run keeps the execution work deadline and records the same allowance in its WorkflowStarted." + }, + "history": [ + { + "id": "01m7er6efbwx1241tprg0q382g", + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "timeout_cleanup": { + "allowance_seconds": 10, + "work_deadlines": { + "execution": "2026-11-13T06:01:00+00:00" + } + } + }, + "recorded_at": "2026-11-13T06:00:01.000000Z" + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "start_timer", + "delay_seconds": 3600 + } + ] + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-outcome-recorded.json b/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-outcome-recorded.json new file mode 100644 index 000000000..7eb9a575e --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-outcome-recorded.json @@ -0,0 +1,56 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "timeout-cleanup-outcome-recorded", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestTimeoutCleanupWorkflow", + "arguments": [ + "complete" + ], + "payload_codec": "avro" + }, + "timeout_cleanup_outcome": { + "issue": "durable-workflow/workflow#863", + "execution_timeout_seconds": 70, + "timeout_cleanup_seconds": 10, + "outcomes": { + "complete": "completed", + "fail": "failed", + "expire": "expired" + }, + "note": "An opted-in run idle past its work deadline is woken, enters timeout cleanup once (WorkflowTimeoutCleanupStarted), receives the timeout interruption, and ends WorkflowTimedOut with the timeout_cleanup outcome. The admitted allowance replays from WorkflowStarted unchanged." + }, + "history": [ + { + "id": "01m7ehvj4edf10ymbdcjs1rk9x", + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "timeout_cleanup": { + "allowance_seconds": 10, + "work_deadlines": { + "execution": "2026-11-13T06:01:00+00:00" + } + } + }, + "recorded_at": "2026-11-13T06:00:00.000000Z" + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "start_timer", + "delay_seconds": 3600 + } + ] + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-redrive-inheritance.json b/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-redrive-inheritance.json new file mode 100644 index 000000000..18ec14c0a --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-redrive-inheritance.json @@ -0,0 +1,50 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "timeout-cleanup-redrive-inheritance", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestTimeoutCleanupWorkflow", + "arguments": [ + "redrive" + ], + "payload_codec": "avro" + }, + "timeout_cleanup_inheritance": { + "issue": "durable-workflow/workflow#863", + "path": "redrive", + "note": "A redrive successor keeps the execution work deadline and records the same allowance in its WorkflowStarted." + }, + "history": [ + { + "id": "01m7e62jfdbvpm1yjy5wjq1nzq", + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "timeout_cleanup": { + "allowance_seconds": 10, + "work_deadlines": { + "execution": "2026-11-13T06:01:00+00:00" + } + } + }, + "recorded_at": "2026-11-13T06:00:02.000000Z" + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "start_timer", + "delay_seconds": 3600 + } + ] + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-task-expiry.json b/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-task-expiry.json new file mode 100644 index 000000000..e92885a98 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/timeout-cleanup-task-expiry.json @@ -0,0 +1,53 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "timeout-cleanup-task-expiry", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestTimeoutCleanupWorkflow", + "arguments": [ + "expire" + ], + "payload_codec": "avro" + }, + "history": [ + { + "id": "01m7ehvj4edf10ymbdcjs1rk9x", + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "timeout_cleanup": { + "allowance_seconds": 10, + "work_deadlines": { + "execution": "2026-11-13T06:01:00+00:00" + } + } + }, + "recorded_at": "2026-11-13T06:00:00.000000Z" + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "start_timer", + "delay_seconds": 3600 + } + ] + }, + "timeout_cleanup_task_expiry": { + "issue": "durable-workflow/workflow#863", + "execution_timeout_seconds": 70, + "timeout_cleanup_seconds": 10, + "detect_after_seconds": 61, + "outcome": "expired", + "note": "A replacement embedded workflow task observes the hard deadline after cleanup began. The terminal history retains the expired cleanup outcome and original cap, matching the watchdog path." + } +} diff --git a/tests/Fixtures/V2/TestTimeoutCleanupWorkflow.php b/tests/Fixtures/V2/TestTimeoutCleanupWorkflow.php new file mode 100644 index 000000000..1aea465cb --- /dev/null +++ b/tests/Fixtures/V2/TestTimeoutCleanupWorkflow.php @@ -0,0 +1,46 @@ +cause = $interruption->cancellation?->cause; + + return match ($mode) { + 'fail' => throw new \RuntimeException('Compensation failed.'), + 'rethrow' => throw $interruption, + 'version-mismatch' => throw new VersionRangeMismatchException('cleanup', 1, 2, 3), + 'history-mismatch' => throw new HistoryEventShapeMismatchException(1, 'activity', ['TimerStarted']), + 'unresolved-failure' => throw UnresolvedWorkflowFailureException::unresolved([ + 'class' => 'UnavailableCleanupFailure', + ]), + 'expire' => (string) timer(3600), + default => 'cleaned-up', + }; + } + } +} diff --git a/tests/Unit/V2/CancellationContextCauseTest.php b/tests/Unit/V2/CancellationContextCauseTest.php new file mode 100644 index 000000000..6db954c29 --- /dev/null +++ b/tests/Unit/V2/CancellationContextCauseTest.php @@ -0,0 +1,115 @@ +assertFalse($context->isTimeout()); + $this->assertSame($snapshot, $context->toArray()); + $this->assertSame( + WorkflowCancellationRequestedException::class, + WorkflowCancellationRequestedException::forContext($context)::class, + ); + // An explicit cancellation cause round-trips with exactly its own keys. + $explicit = [ + ...$snapshot, + 'cause' => 'cancellation', + ]; + $this->assertFalse(CancellationContext::fromArray($explicit)->isTimeout()); + $this->assertSame($explicit, CancellationContext::fromArray($explicit)->toArray()); + } + + public function testATimeoutCauseRoundTripsAndDeliversTheTimeoutSubtype(): void + { + $context = CancellationContext::fromArray([ + ...self::snapshot(), + 'cause' => 'timeout', + ]); + + $this->assertTrue($context->isTimeout()); + $this->assertSame('timeout', $context->toArray()['cause']); + $exception = WorkflowCancellationRequestedException::forContext($context); + $this->assertInstanceOf(WorkflowTimeoutCleanupRequested::class, $exception); + $this->assertInstanceOf(WorkflowCancellationRequestedException::class, $exception); + $this->assertSame($context, $exception->cancellation); + } + + public function testADescendantIsCancelledNotTimedOut(): void + { + $context = CancellationContext::fromArray([ + ...self::snapshot(), + 'cause' => 'timeout', + ]); + + $this->assertFalse($context->forDescendant('req-child', 'child-instance', 'child-run')->isTimeout()); + } + + public function testAScopedDescendantOfATimeoutIsCancelledNotTimedOut(): void + { + $root = CancellationContext::fromArray([ + ...self::snapshot(), + 'cause' => 'timeout', + ]); + $origin = \Workflow\V2\ScopedCancellationContext::fromRunContext($root); + $child = CancellationContext::fromScopeContext($origin, 'req-child', 'child-instance', 'child-run'); + + $this->assertFalse($child->isTimeout()); + $this->assertArrayNotHasKey('cause', $child->toArray()); + // The child's snapshot round-trips: it is not required to carry the root's cause. + $this->assertFalse(CancellationContext::fromArray($child->toArray())->isTimeout()); + } + + public function testAnUnknownCauseIsRefused(): void + { + $this->expectException(InvalidArgumentException::class); + + CancellationContext::fromArray([ + ...self::snapshot(), + 'cause' => 'deadline', + ]); + } + + /** + * @return array + */ + private static function snapshot(): array + { + return [ + 'schema' => 'durable-workflow.cancellation-context/v1', + 'request_id' => 'req-1', + 'root_request_id' => 'req-1', + 'root_workflow_instance_id' => 'instance-1', + 'root_workflow_run_id' => 'run-1', + 'parent_request_id' => null, + 'reason' => 'work deadline passed', + 'requester' => [ + 'type' => 'system', + 'id' => 'timeout_cleanup', + ], + 'source' => 'system', + 'requested_at' => '2026-11-13T00:01:00.000000Z', + 'cleanup_deadline_at' => '2026-11-13T00:01:10.000000Z', + 'lineage' => [[ + 'request_id' => 'req-1', + 'workflow_instance_id' => 'instance-1', + 'workflow_run_id' => 'run-1', + ]], + ]; + } +} diff --git a/tests/Unit/V2/ChildClosureSiblingHistoryCostTest.php b/tests/Unit/V2/ChildClosureSiblingHistoryCostTest.php index a0ea56ad9..6c2431113 100644 --- a/tests/Unit/V2/ChildClosureSiblingHistoryCostTest.php +++ b/tests/Unit/V2/ChildClosureSiblingHistoryCostTest.php @@ -46,6 +46,8 @@ final class ChildClosureSiblingHistoryCostTest extends TestCase protected function setUp(): void { parent::setUp(); + // Periodic heartbeat/pruning writes are independent of sibling count. + $this->travelTo(now()->startOfSecond()); Queue::fake(); config() ->set('queue.default', 'redis'); diff --git a/tests/Unit/V2/StartOptionsTimeoutCleanupTest.php b/tests/Unit/V2/StartOptionsTimeoutCleanupTest.php new file mode 100644 index 000000000..b6152d363 --- /dev/null +++ b/tests/Unit/V2/StartOptionsTimeoutCleanupTest.php @@ -0,0 +1,69 @@ +withTimeoutCleanup(10); + + $this->assertSame(10, $options->timeoutCleanupSeconds); + $this->assertSame(70, $options->executionTimeoutSeconds); + $this->assertSame(40, $options->runTimeoutSeconds); + + $start = Carbon::parse('2026-11-13T00:00:00Z'); + $this->assertSame([ + 'allowance_seconds' => 10, + 'work_deadlines' => [ + 'execution' => '2026-11-13T00:01:00+00:00', + 'run' => '2026-11-13T00:00:30+00:00', + ], + ], TimeoutCleanup::startedPayload(10, $start->copy()->addSeconds(70), $start->copy()->addSeconds(40))); + $this->assertNull(TimeoutCleanup::startedPayload(null, $start->copy()->addSeconds(70), null)); + } + + public function testWithoutTheOptionNothingChanges(): void + { + $this->assertNull((new StartOptions(executionTimeoutSeconds: 70))->timeoutCleanupSeconds); + } + + /** + * @return iterable + */ + public static function refusals(): iterable + { + yield 'no hard timeout' => [null, null, 10]; + yield 'not shorter than the execution timeout' => [10, null, 10]; + yield 'not shorter than the run timeout' => [70, 10, 10]; + yield 'zero' => [70, null, 0]; + yield 'over a day' => [100000, null, 86401]; + } + + #[DataProvider('refusals')] + public function testAnAllowanceThatLeavesNoWorkTimeIsRefusedAtAdmission( + ?int $execution, + ?int $run, + int $cleanup, + ): void { + $this->expectException(LogicException::class); + + new StartOptions( + executionTimeoutSeconds: $execution, + runTimeoutSeconds: $run, + timeoutCleanupSeconds: $cleanup, + ); + } +}