diff --git a/CHANGELOG.md b/CHANGELOG.md index b8f7185f..d979fcca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,26 @@ # Changelog +## Unreleased + +- Bound `record_local_activity` timing fields (`duration_ms`, heartbeat + `elapsed_ms`, `backoff_seconds`) and refuse a report whose walk back from + completion leaves the storable timestamp range (#930). +- Meter each workflow task completion with hard structural limits on history + events, history bytes, metered rows and externalized payload bytes + (`history_transaction_size`, `history_transaction_bytes`, + `history_transaction_rows`, `external_payload_transaction_bytes`). Metered + rows are history events and local activity records for now; other command + writes are not counted yet. A refusal rolls the whole completion back and + keeps the task's lease. A run's deadline close is never metered. The limits + appear in the structural limit snapshot. +- Dispatch lifecycle events from completions only after the completion + commits, so a refused or retried completion dispatches none. +- Add the optional `PredictableExternalPayloadStorageDriver` interface. A + driver that implements it has payloads staged and stored only after the + completion passes every limit. The built-in local driver implements it, and + now stores objects atomically after checking their length and hash. Other + drivers, including the filesystem driver, keep storing payloads as before. + ## 2.13.1 - 2026-10-24 - Cancel the live task when a retried activity receives cooperative cancellation. diff --git a/docs/architecture/completion-budget.md b/docs/architecture/completion-budget.md new file mode 100644 index 00000000..2b4d961f --- /dev/null +++ b/docs/architecture/completion-budget.md @@ -0,0 +1,183 @@ +# Workflow Task Completion Budget + +A workflow task completion is one database transaction. It applies a worker's +commands and everything they cause. The completion budget bounds that +transaction, so a valid completion can neither drive timestamp arithmetic out +of range nor write an unbounded amount of history or external payload data. +Rows are bounded for the write sites listed under Phases; the rest follow in +Phase B. + +## Timing fields of `record_local_activity` + +Local activity reports carry elapsed times, not timestamps. Recording walks +each attempt back from the completion instant by its duration and the +following retry backoff. The normalizer bounds every field it walks, with an +ordinary 422 field error: + +| Field | Maximum | Constant | +| --- | --- | --- | +| `attempts.*.duration_ms` | 604,800,000 (7 days) | `WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_DURATION_MS` | +| `attempts.*.heartbeats.*.elapsed_ms` | 604,800,000 (7 days) | `WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_DURATION_MS` | +| `attempts.*.backoff_seconds` | 86,400 (1 day) | `WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_BACKOFF_SECONDS` | + +With at most 100 attempts, a report reaches back at most about 800 days. The +bridge also sums the walk with overflow-checked arithmetic. Before writing, it +refuses with `local_activity_timestamp_out_of_range` if the earliest derived +timestamp falls before the earliest timestamp every supported column stores +(MySQL `TIMESTAMP`'s 1970-01-01T00:00:01Z). + +## The early check + +After a run's deadline close (see below) and before the worker's first write, +`complete()` counts the history events and rows the completion's local +activities will write. The count is exact: per command, 1 scheduled event, +then per attempt 1 started event, its heartbeats and a retry event before the +next attempt, then 1 terminal event. Rows are two per event (the event and +the run's history position), plus the execution, each attempt and a failure +when the activity failed or timed out. Every other command counts its lower +bound of one event. + +Past a limit, the completion is refused with `completion_budget_exceeded` +before anything is written, and the worker keeps its lease. This refuses the +local-activity amplifier cheaply. It is not the whole budget. + +## The history transaction meter + +`HistoryTransactionMeter` is the authority. `complete()` and the checkpointed +command paths run each transaction attempt's body under a fresh meter. A +retried attempt starts over with an empty stage. + +- **Events and bytes.** `WorkflowHistoryEvent::record()` is the only writer of + history. It charges one event, and its bytes as `HistoryBudget` measures a + stored event: the event type length plus the length of + `json_encode($payload, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE)`. + It also charges two rows: the event and the run position it advances. If the + payload cannot be encoded, the completion fails closed with + `history_transaction_measure_unavailable`. +- **Rows.** A metered write site charges the rows its statement will affect + before the statement runs. Phase A meters every history write and every + local activity row. Other command writes are not counted yet; Phase B + charges them (see below). +- **Refusal.** A charge past a limit throws + `HistoryTransactionLimitExceededException`. That trips the meter: any later + charge, and the end of the body, throw again. Code that catches the + exception therefore cannot continue to a commit. The exception is handled + only outside the transaction callback. The transaction rolls back, so no + history, execution, attempt, failure, projection, task or receipt row + survives, and the task keeps its lease and expiry. The bridge answers + `completed: false` with `reason: history_transaction_limit_exceeded` and + `limit` naming the dimension. +- **Lifecycle events.** Public lifecycle events (workflow completed, failed or + started by continue-as-new, activity failed, failure recorded) report + committed state. Inside a metered transaction they are held back and + dispatched once, after the transaction commits. A refused attempt, a failed + staged put, a failed commit and a retried attempt dispatch none of them. + +## Limits + +All of these are hard `StructuralLimits`. A value of 0 disables that +dimension. `StructuralLimits::snapshot()` publishes each one, so the health +check, operator metrics and backend capability reports show them. + +| Config (`workflows.v2.structural_limits.*`) | Environment | Default | Meaning | +| --- | --- | --- | --- | +| `history_transaction_size` | `DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE` | 5,000 | History events one completion may write | +| `history_transaction_bytes` | `DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES` | 16 MiB | History bytes one completion may write | +| `history_transaction_rows` | `DW_V2_LIMIT_HISTORY_TRANSACTION_ROWS` | 20,000 | Metered rows one completion may write. Phase A counts history events with their run positions, and local activity rows. | +| `external_payload_transaction_bytes` | `DW_V2_LIMIT_EXTERNAL_PAYLOAD_TRANSACTION_BYTES` | 16 MiB | Raw payload bytes one completion may externalize | +| `payload_size_bytes` (existing) | `DW_V2_LIMIT_PAYLOAD_SIZE_BYTES` | 2 MiB | Each raw payload externalized | + +Until Phase B lands, `history_transaction_rows` does not bound the rows of +other command writes or the cascades they trigger. The pending-count limits +bound those. + +`HistoryBudget`'s continue-as-new thresholds stay advisory. They recommend +continue-as-new and never refuse a write. + +## External payloads + +A database rollback cannot undo an object stored in external payload storage. +Every payload a metered completion externalizes is first checked against +`payload_size_bytes` and charged to `external_payload_transaction_bytes`. +Every occurrence counts, even when identical content will be stored once. +What happens next depends on the driver. + +**Drivers that predict their URI** implement +`PredictableExternalPayloadStorageDriver`. Their payloads are staged, and +stored only after the completion is admitted: + +1. `uriFor($sha256, $codec)` returns the URI `put()` will return for that + content. The payload is registered in the meter's stage under its driver + and URI, and the final stored envelope, with the predicted URI, is written + to rows and history. The meter therefore measures the exact bytes that + commit. +2. If the same driver predicts one URI for different content, hash or codec, + the completion is refused with `external_payload_staged_uri_collision`. + Two drivers that predict the same URI keep separate entries. +3. After the body has passed every limit, still inside the transaction and + before commit, the stage stores each object in order. Each `put()` must + return its predicted URI. A mismatch fails with + `external_payload_staged_uri_mismatch`, and any failure rolls back. + +A limit refusal happens before the stage is flushed, so it stores nothing. + +The interface requires `put()` to return exactly `uriFor()`, refuse content +that does not match its SHA-256, never leave a partial or corrupt object at +the URI, and be idempotent. The built-in local driver implements it. It keeps +an existing object only if its length and hash match. Otherwise it writes the +content aside, checks the exact length and hash, and renames it into place. + +**Other drivers** implement only `ExternalPayloadStorageDriver`, as before. +The built-in Laravel filesystem driver is one of them, since a filesystem disk +does not guarantee an atomic write. These drivers keep storing each payload +when it is externalized, after its bytes are charged. If the completion is +then refused or rolled back, the objects it stored stay, unreferenced. The +exposure is bounded by `external_payload_transaction_bytes` per transaction +attempt. To store nothing on refusal, implement `uriFor()` and the guarantees +above. + +**Orphans.** If a stage flush or commit fails after some objects are stored, +those objects stay. Content-addressed objects may be shared, so a rollback +never deletes them. Reading a payload checks its size and hash against the +reference, and reachability-based retention collects unreferenced objects. + +## A run's deadline close + +When a worker completes a task whose run is past its execution, run or +cleanup deadline, `complete()` closes the run before applying any command. No +other path closes a run while its workflow task is open. That close is the +server's own work on stored state, not the worker's, so it runs unmetered +(`HistoryTransactionMeter::unmetered()`) and before the early check. A run +whose close is larger than one completion's budget still closes, instead of +refusing every completion. + +## Phases + +**Phase A** (this change) covers: + +- the timing caps and the storage-floor guard; +- the exact local-activity early check; +- the writer-level event and byte meter on every path, checkpoints included; +- history and local-activity row charges; +- the raw external payload byte limits, with staging for drivers that predict + their URI; +- lifecycle events held back until commit; +- the unmetered deadline close. + +**Phase B** charges every remaining write site before its statement runs: + +- the pending-signal and pending-update closers; +- the continue-as-new handoffs; +- parent-close enforcement; +- selection-cancel leaves; +- service operations; +- projections; +- memo and search-attribute writes. + +Each bulk site reads at most the remaining rows plus one before it decides. +The row limit bounds a whole completion only when Phase B lands. + +Phase B also depends on three separate fixes: service operations taking their +namespace and caller identity from the server instead of the worker, no +sleeps inside the completion transaction, and pending-count limits on +external-worker completions. diff --git a/src/V2/Contracts/PredictableExternalPayloadStorageDriver.php b/src/V2/Contracts/PredictableExternalPayloadStorageDriver.php new file mode 100644 index 00000000..8884e6ea --- /dev/null +++ b/src/V2/Contracts/PredictableExternalPayloadStorageDriver.php @@ -0,0 +1,31 @@ +findOrFail($run->id); $now = now(); $sequence = self::nextSequenceForRun($lockedRun); + $storedPayload = self::snapshotPayload($payload, $taskModel, $commandModel); + // A metered completion (#930) charges the event as stored, + // and the run row whose history position it advances. + $meter = HistoryTransactionMeter::current(); + $meter?->chargeEvent($eventType->value, $storedPayload); + $meter?->chargeRows(1); /** @var self $event */ $event = ConfiguredV2Models::query('history_event_model', self::class)->create([ 'workflow_run_id' => $lockedRun->id, 'sequence' => $sequence, 'event_type' => $eventType->value, - 'payload' => self::snapshotPayload($payload, $taskModel, $commandModel), + 'payload' => $storedPayload, 'workflow_task_id' => $taskId, 'workflow_command_id' => $commandId, 'recorded_at' => $now, diff --git a/src/V2/Support/DefaultWorkflowTaskBridge.php b/src/V2/Support/DefaultWorkflowTaskBridge.php index 69e3e659..126788ee 100644 --- a/src/V2/Support/DefaultWorkflowTaskBridge.php +++ b/src/V2/Support/DefaultWorkflowTaskBridge.php @@ -39,6 +39,7 @@ use Workflow\V2\Enums\UpdateStatus; use Workflow\V2\Exceptions\ActivityTimeoutException; use Workflow\V2\Exceptions\HistoryEventShapeMismatchException; +use Workflow\V2\Exceptions\HistoryTransactionLimitExceededException; use Workflow\V2\Models\ActivityAttempt; use Workflow\V2\Models\ActivityExecution; use Workflow\V2\Models\WorkflowChildCall; @@ -61,6 +62,12 @@ final class DefaultWorkflowTaskBridge implements CooperativeWorkflowTaskBridge, public const AVAILABILITY_CEILING_SECONDS = 1; + /** + * The earliest timestamp every supported column type stores (MySQL + * TIMESTAMP's 1970-01-01T00:00:01Z), in Unix milliseconds. + */ + private const EARLIEST_STORED_TIMESTAMP_MS = 1000; + /** * Relations the history-projection role consumes for a workflow-bridge * projection. Hydrated by {@see self::projectRun()} before the role is @@ -1234,200 +1241,237 @@ public function complete(string $taskId, array $commands): array ]; } - return DB::transaction(function () use ($taskId, $parsed): array { - /** @var WorkflowTask|null $task */ - $task = ConfiguredV2Models::query('task_model', WorkflowTask::class) - ->lockForUpdate() - ->find($taskId); - - if ($parsed === null && ($task === null || ! self::isSignalResumeTask($task))) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $task?->workflow_run_id, - 'run_status' => null, - 'created_task_ids' => [], - 'reason' => 'invalid_commands', - ]; - } + // The whole completion runs under one history transaction meter + // (#930): every event, byte, row and staged external payload is + // charged as it is written, and a refusal rolls everything back, + // leaving the task's lease as it was. + try { + return self::meteredTransaction(function () use ($taskId, $parsed, $commands): array { + /** @var WorkflowTask|null $task */ + $task = ConfiguredV2Models::query('task_model', WorkflowTask::class) + ->lockForUpdate() + ->find($taskId); - if ($task === null || $task->task_type !== TaskType::Workflow) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => null, - 'run_status' => null, - 'created_task_ids' => [], - 'reason' => $task === null ? 'task_not_found' : 'task_not_workflow', - ]; - } + if ($parsed === null && ($task === null || ! self::isSignalResumeTask($task))) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $task?->workflow_run_id, + 'run_status' => null, + 'created_task_ids' => [], + 'reason' => 'invalid_commands', + ]; + } - $parsed ??= [ - 'non_terminal' => [], - 'terminal' => null, - ]; + if ($task === null || $task->task_type !== TaskType::Workflow) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => null, + 'run_status' => null, + 'created_task_ids' => [], + 'reason' => $task === null ? 'task_not_found' : 'task_not_workflow', + ]; + } - if ($task->status !== TaskStatus::Leased) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $task->workflow_run_id, - 'run_status' => null, - 'created_task_ids' => [], - 'reason' => 'task_not_leased', + $parsed ??= [ + 'non_terminal' => [], + 'terminal' => null, ]; - } - /** @var WorkflowRun|null $run */ - $run = ConfiguredV2Models::query('run_model', WorkflowRun::class) - ->lockForUpdate() - ->find($task->workflow_run_id); + if ($task->status !== TaskStatus::Leased) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $task->workflow_run_id, + 'run_status' => null, + 'created_task_ids' => [], + 'reason' => 'task_not_leased', + ]; + } - if ($run === null) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $task->workflow_run_id, - 'run_status' => null, - 'created_task_ids' => [], - 'reason' => 'run_not_found', - ]; - } + /** @var WorkflowRun|null $run */ + $run = ConfiguredV2Models::query('run_model', WorkflowRun::class) + ->lockForUpdate() + ->find($task->workflow_run_id); + + if ($run === null) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $task->workflow_run_id, + 'run_status' => null, + 'created_task_ids' => [], + 'reason' => 'run_not_found', + ]; + } - if ($run->status->isTerminal()) { - $task->forceFill([ - 'status' => $run->status === RunStatus::Failed ? TaskStatus::Failed : TaskStatus::Completed, - 'lease_expires_at' => null, - ])->save(); + if ($run->status->isTerminal()) { + $task->forceFill([ + 'status' => $run->status === RunStatus::Failed ? TaskStatus::Failed : TaskStatus::Completed, + 'lease_expires_at' => null, + ])->save(); - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'run_status' => $run->status->value, - 'created_task_ids' => [], - 'reason' => 'run_already_closed', - ]; - } + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => 'run_already_closed', + ]; + } - if ($this->executor->timeoutIfDeadlineExpired($run, $task)) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'run_status' => $run->status->value, - 'created_task_ids' => [], - 'reason' => 'run_timed_out', - ]; - } + // A run's deadline close is the server's own work on stored + // state, not the worker's commands, and nothing else closes + // a run whose task is open: it is never metered (#930). + if (HistoryTransactionMeter::unmetered( + fn (): bool => $this->executor->timeoutIfDeadlineExpired($run, $task) + )) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => 'run_timed_out', + ]; + } - if ($this->executor->cancelIfCleanupDeadlineExpired($run, $task)) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'run_status' => $run->status->value, - 'created_task_ids' => [], - 'reason' => 'run_cancelled', - ]; - } + if (HistoryTransactionMeter::unmetered( + fn (): bool => $this->executor->cancelIfCleanupDeadlineExpired($run, $task) + )) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => 'run_cancelled', + ]; + } - $sequence = WorkflowStepHistory::nextDurableCommandSequence($run); - $createdTaskIds = []; - $scopeError = $this->validateCancellationScopeMembership($run, $parsed['non_terminal'], $sequence); - if ($scopeError !== null) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'run_status' => $run->status->value, - 'created_task_ids' => [], - 'reason' => $scopeError, - ]; - } - $invalidUpdateCommands = $this->validateUpdateCommands($run, $task, $parsed['non_terminal']); + // The early whole-completion check (#930): after any deadline + // close, before the worker's first write; the lease stays. + if (WorkflowCommandNormalizer::completionBudgetErrors(array_values($commands)) !== []) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => null, + 'created_task_ids' => [], + 'reason' => 'completion_budget_exceeded', + ]; + } - if ($invalidUpdateCommands !== null) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'run_status' => $run->status->value, - 'created_task_ids' => [], - 'reason' => $invalidUpdateCommands, - ]; - } + $sequence = WorkflowStepHistory::nextDurableCommandSequence($run); + $createdTaskIds = []; + $scopeError = $this->validateCancellationScopeMembership($run, $parsed['non_terminal'], $sequence); + if ($scopeError !== null) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => $scopeError, + ]; + } + $invalidUpdateCommands = $this->validateUpdateCommands($run, $task, $parsed['non_terminal']); + + if ($invalidUpdateCommands !== null) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => $invalidUpdateCommands, + ]; + } - if (! $this->selectionCancellationCommandsAreValid($run, $parsed['non_terminal'])) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'run_status' => $run->status->value, - 'created_task_ids' => [], - 'reason' => 'invalid_commands', - ]; - } + if (! $this->selectionCancellationCommandsAreValid($run, $parsed['non_terminal'])) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => 'invalid_commands', + ]; + } - if (! self::parallelCommandsMatchSequences($parsed['non_terminal'], $sequence, $run)) { - return [ - 'completed' => false, - 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'run_status' => $run->status->value, - 'created_task_ids' => [], - 'reason' => 'invalid_commands', - ]; - } + if (! self::parallelCommandsMatchSequences($parsed['non_terminal'], $sequence, $run)) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => 'invalid_commands', + ]; + } - $this->recordAppliedSignalForSignalResume($run, $task); - $conditionSelectionResolution = $this->recordSatisfiedConditionWaitForSignalResume( - $run, - $task, - $parsed['non_terminal'], - ); + $this->recordAppliedSignalForSignalResume($run, $task); + $conditionSelectionResolution = $this->recordSatisfiedConditionWaitForSignalResume( + $run, + $task, + $parsed['non_terminal'], + ); - foreach ($parsed['non_terminal'] as $command) { - $sequence = $this->applyNonTerminalCommand($run, $task, $command, $sequence, $createdTaskIds); - } + foreach ($parsed['non_terminal'] as $command) { + $sequence = $this->applyNonTerminalCommand($run, $task, $command, $sequence, $createdTaskIds); + } - $terminal = $parsed['terminal']; + $terminal = $parsed['terminal']; - // A signal may commit after the worker has read history. Give it - // an ordered delivery turn before accepting that worker's close. - if (in_array($terminal['type'] ?? null, ['complete_workflow', 'continue_as_new'], true) - && ! is_string($run->cancellation_request_command_id) - && PendingMessageTask::hasDeliverableSignal($run, self::workflowSignalIdForTask($task))) { - $terminal = null; - } + // A signal may commit after the worker has read history. Give it + // an ordered delivery turn before accepting that worker's close. + if (in_array($terminal['type'] ?? null, ['complete_workflow', 'continue_as_new'], true) + && ! is_string($run->cancellation_request_command_id) + && PendingMessageTask::hasDeliverableSignal($run, self::workflowSignalIdForTask($task))) { + $terminal = null; + } - if ($terminal !== null) { - if ($terminal['type'] === 'continue_as_new') { - $this->applyContinueAsNew($run, $task, $sequence, $terminal, $createdTaskIds); - } elseif ($terminal['type'] === 'complete_workflow') { - $this->applyWorkflowCompletion($run, $task, $terminal); + if ($terminal !== null) { + if ($terminal['type'] === 'continue_as_new') { + $this->applyContinueAsNew($run, $task, $sequence, $terminal, $createdTaskIds); + } elseif ($terminal['type'] === 'complete_workflow') { + $this->applyWorkflowCompletion($run, $task, $terminal); + } else { + $this->applyWorkflowFailure($run, $task, $terminal); + } } else { - $this->applyWorkflowFailure($run, $task, $terminal); + $this->markRunWaiting( + $run, + $task, + $parsed['non_terminal'], + $createdTaskIds, + $conditionSelectionResolution, + ); } - } else { - $this->markRunWaiting( - $run, - $task, - $parsed['non_terminal'], - $createdTaskIds, - $conditionSelectionResolution, - ); - } + return [ + 'completed' => true, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => $createdTaskIds, + 'reason' => null, + ]; + }); + } catch (HistoryTransactionLimitExceededException $refusal) { return [ - 'completed' => true, + 'completed' => false, 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'run_status' => $run->status->value, - 'created_task_ids' => $createdTaskIds, - 'reason' => null, + 'workflow_run_id' => null, + 'run_status' => null, + 'created_task_ids' => [], + 'reason' => self::meterRefusalReason($refusal), + 'limit' => $refusal->dimension, + 'message' => $refusal->getMessage(), ]; - }); + } } /** @@ -1470,9 +1514,43 @@ public function validateCancellationScopeMembership(WorkflowRun $run, array $com return null; } - /** @param list $commands - * @return array + /** + * Run one metered transaction (#930): each attempt gets a fresh history + * transaction meter, and the lifecycle events its body dispatches are + * held back until the transaction has committed, so a refused, failed or + * retried attempt dispatches none. + * + * @template T + * @param callable(HistoryTransactionMeter): T $body + * @return T */ + private static function meteredTransaction(callable $body, int $attempts = 1): mixed + { + [$result, $events] = DB::transaction( + static fn (): array => LifecycleEventDispatcher::collect( + static fn (): mixed => HistoryTransactionMeter::run($body) + ), + $attempts, + ); + LifecycleEventDispatcher::release($events); + + return $result; + } + + /** + * The completion refusal reason for a history transaction meter refusal: + * its own reason, or history_transaction_limit_exceeded for a limit. + */ + private static function meterRefusalReason(HistoryTransactionLimitExceededException $refusal): string + { + return in_array($refusal->dimension, [ + HistoryTransactionLimitExceededException::MEASURE_UNAVAILABLE, + HistoryTransactionLimitExceededException::STAGED_URI_MISMATCH, + HistoryTransactionLimitExceededException::STAGED_URI_COLLISION, + HistoryTransactionLimitExceededException::TIMESTAMP_OUT_OF_RANGE, + ], true) ? $refusal->dimension : 'history_transaction_limit_exceeded'; + } + private function checkpointCommands( string $taskId, string $leaseOwner, @@ -1534,205 +1612,211 @@ private function checkpointCommands( return $refused('task_not_found'); } - return DB::transaction(function () use ( - $snapshot, - $taskId, - $leaseOwner, - $workflowTaskAttempt, - $checkpointId, - $startSequence, - $parsed, - $fingerprint, - $refused, - $group, - $scopePrefix, - $checkpointName, - ): array { - /** @var WorkflowRun|null $run */ - $run = ConfiguredV2Models::query('run_model', WorkflowRun::class) - ->lockForUpdate() - ->find($snapshot->workflow_run_id); - /** @var WorkflowTask|null $task */ - $task = ConfiguredV2Models::query('task_model', WorkflowTask::class)->lockForUpdate()->find($taskId); - if ($run === null || $task === null || $task->workflow_run_id !== $run->id) { - return $refused('workflow_claim_not_found'); - } - if ($task->task_type !== TaskType::Workflow || $task->status !== TaskStatus::Leased - || $task->lease_owner !== $leaseOwner || $task->attempt_count !== $workflowTaskAttempt) { - return $refused('workflow_claim_mismatch'); - } - if ($task->lease_expires_at === null || now()->gte($task->lease_expires_at)) { - return $refused('workflow_claim_expired'); - } - if ($run->status->isTerminal()) { - return $refused('run_closed'); - } - if (($run->execution_deadline_at !== null && now()->gte($run->execution_deadline_at)) - || ($run->run_deadline_at !== null && now()->gte($run->run_deadline_at))) { - return $refused('run_deadline_expired'); - } - $payload = is_array($task->payload) ? $task->payload : []; - $receiptKey = $scopePrefix ? 'portable_scope_checkpoint' - : ($group ? 'portable_local_group_checkpoint' : 'portable_local_checkpoint'); - $receipt = $payload[$receiptKey] ?? null; - if (is_array($receipt) && ($receipt['checkpoint_id'] ?? null) === $checkpointId) { - if (($receipt['workflow_task_attempt'] ?? null) !== $workflowTaskAttempt - || ($receipt['lease_owner'] ?? null) !== $leaseOwner - || ($receipt['start_sequence'] ?? null) !== $startSequence - || ($receipt['fingerprint'] ?? null) !== $fingerprint) { - return $refused($checkpointName . '_checkpoint_mismatch'); + // Checkpointed commands are metered like a completion (#930); each + // retried transaction attempt starts a fresh meter and stage. + try { + return self::meteredTransaction(function () use ( + $snapshot, + $taskId, + $leaseOwner, + $workflowTaskAttempt, + $checkpointId, + $startSequence, + $parsed, + $fingerprint, + $refused, + $group, + $scopePrefix, + $checkpointName, + ): array { + /** @var WorkflowRun|null $run */ + $run = ConfiguredV2Models::query('run_model', WorkflowRun::class) + ->lockForUpdate() + ->find($snapshot->workflow_run_id); + /** @var WorkflowTask|null $task */ + $task = ConfiguredV2Models::query('task_model', WorkflowTask::class)->lockForUpdate()->find($taskId); + if ($run === null || $task === null || $task->workflow_run_id !== $run->id) { + return $refused('workflow_claim_not_found'); } - // This receipt proves prefix commit, not permission to start - // application code. Preparation still checks cancellation. - return [ - ...$receipt, - 'checkpointed' => true, - 'duplicate' => true, - 'reason' => null, - ]; - } - if ($run->cancellation_request_command_id !== null) { - $delivery = $run->historyEvents() - ->where('event_type', HistoryEventType::CooperativeCancellationDelivered) - ->where('workflow_command_id', $run->cancellation_request_command_id) - ->first(); - $cleanup = PortableLocalActivityCleanup::snapshot($run, [ - 'request_id' => $run->cancellation_request_command_id, - 'delivery_history_event_id' => $delivery?->id, - ], $startSequence); - if ($cleanup === null) { - // An atomic local group can prove scoped cleanup on every - // member. Ordinary prefixes still need root delivery. - if (! $group || $parsed['non_terminal'] === []) { - return $refused('cancellation_requested'); + if ($task->task_type !== TaskType::Workflow || $task->status !== TaskStatus::Leased + || $task->lease_owner !== $leaseOwner || $task->attempt_count !== $workflowTaskAttempt) { + return $refused('workflow_claim_mismatch'); + } + if ($task->lease_expires_at === null || now()->gte($task->lease_expires_at)) { + return $refused('workflow_claim_expired'); + } + if ($run->status->isTerminal()) { + return $refused('run_closed'); + } + if (($run->execution_deadline_at !== null && now()->gte($run->execution_deadline_at)) + || ($run->run_deadline_at !== null && now()->gte($run->run_deadline_at))) { + return $refused('run_deadline_expired'); + } + $payload = is_array($task->payload) ? $task->payload : []; + $receiptKey = $scopePrefix ? 'portable_scope_checkpoint' + : ($group ? 'portable_local_group_checkpoint' : 'portable_local_checkpoint'); + $receipt = $payload[$receiptKey] ?? null; + if (is_array($receipt) && ($receipt['checkpoint_id'] ?? null) === $checkpointId) { + if (($receipt['workflow_task_attempt'] ?? null) !== $workflowTaskAttempt + || ($receipt['lease_owner'] ?? null) !== $leaseOwner + || ($receipt['start_sequence'] ?? null) !== $startSequence + || ($receipt['fingerprint'] ?? null) !== $fingerprint) { + return $refused($checkpointName . '_checkpoint_mismatch'); } - foreach ($parsed['non_terminal'] as $offset => $command) { - if ($command['type'] !== 'prepare_local_activity') { + // This receipt proves prefix commit, not permission to start + // application code. Preparation still checks cancellation. + return [ + ...$receipt, + 'checkpointed' => true, + 'duplicate' => true, + 'reason' => null, + ]; + } + if ($run->cancellation_request_command_id !== null) { + $delivery = $run->historyEvents() + ->where('event_type', HistoryEventType::CooperativeCancellationDelivered) + ->where('workflow_command_id', $run->cancellation_request_command_id) + ->first(); + $cleanup = PortableLocalActivityCleanup::snapshot($run, [ + 'request_id' => $run->cancellation_request_command_id, + 'delivery_history_event_id' => $delivery?->id, + ], $startSequence); + if ($cleanup === null) { + // An atomic local group can prove scoped cleanup on every + // member. Ordinary prefixes still need root delivery. + if (! $group || $parsed['non_terminal'] === []) { return $refused('cancellation_requested'); } - $memberCleanup = PortableLocalActivityCleanup::snapshot( - $run, - $command['cancellation_cleanup'] ?? null, - $startSequence + $offset, - $command['cancellation_scope_id'] ?? CancellationScopeHistory::ROOT_SCOPE_ID, - ); - if (! PortableLocalActivityCleanup::belongsToRunRequest($run, $memberCleanup)) { - return $refused('cancellation_requested'); + foreach ($parsed['non_terminal'] as $offset => $command) { + if ($command['type'] !== 'prepare_local_activity') { + return $refused('cancellation_requested'); + } + $memberCleanup = PortableLocalActivityCleanup::snapshot( + $run, + $command['cancellation_cleanup'] ?? null, + $startSequence + $offset, + $command['cancellation_scope_id'] ?? CancellationScopeHistory::ROOT_SCOPE_ID, + ); + if (! PortableLocalActivityCleanup::belongsToRunRequest($run, $memberCleanup)) { + return $refused('cancellation_requested'); + } } } + if (now()->gte($run->cancellation_deadline_at)) { + return $refused('cancellation_deadline_expired'); + } } - if (now()->gte($run->cancellation_deadline_at)) { - return $refused('cancellation_deadline_expired'); - } - } - $sequence = WorkflowStepHistory::nextDurableCommandSequence($run); - if ($sequence !== $startSequence) { - return $refused($checkpointName . '_checkpoint_sequence_mismatch'); - } - $scopeError = $this->validateCancellationScopeMembership($run, $parsed['non_terminal'], $sequence); - if ($scopeError !== null) { - return $refused($scopeError); - } - $invalidUpdate = $this->validateUpdateCommands($run, $task, $parsed['non_terminal']); - if ($invalidUpdate !== null) { - return $refused($invalidUpdate); - } - if (! self::parallelCommandsMatchSequences($parsed['non_terminal'], $sequence, $run)) { - return $refused('invalid_' . $checkpointName . '_checkpoint_commands'); - } - // Validate every local scope/cleanup member before creating any sibling. - foreach ($parsed['non_terminal'] as $offset => $command) { - if ($command['type'] !== 'prepare_local_activity') { - continue; + $sequence = WorkflowStepHistory::nextDurableCommandSequence($run); + if ($sequence !== $startSequence) { + return $refused($checkpointName . '_checkpoint_sequence_mismatch'); } - if (! CancellationScopeHistory::isRecordedBefore( - $run, - $command['cancellation_scope_id'] ?? CancellationScopeHistory::ROOT_SCOPE_ID, - $sequence + $offset, - )) { - return $refused('local_activity_scope_not_recorded'); + $scopeError = $this->validateCancellationScopeMembership($run, $parsed['non_terminal'], $sequence); + if ($scopeError !== null) { + return $refused($scopeError); } - $cleanup = PortableLocalActivityCleanup::snapshot( - $run, - $command['cancellation_cleanup'] ?? null, - $sequence + $offset, - $command['cancellation_scope_id'] ?? CancellationScopeHistory::ROOT_SCOPE_ID, - ); - if ((isset($command['cancellation_cleanup']) && $cleanup === null) - || ($run->cancellation_request_command_id !== null - && ! PortableLocalActivityCleanup::belongsToRunRequest($run, $cleanup)) - || ($run->cancellation_request_command_id === null && isset($command['cancellation_cleanup']) && ! isset($cleanup['scope_id']))) { - return $refused('local_activity_cleanup_authority_mismatch'); + $invalidUpdate = $this->validateUpdateCommands($run, $task, $parsed['non_terminal']); + if ($invalidUpdate !== null) { + return $refused($invalidUpdate); } - if ($cleanup !== null && now()->gte(PortableLocalActivityCleanup::snapshotDeadline($cleanup))) { - return $refused(isset($cleanup['scope_id']) ? 'local_activity_cleanup_deadline_expired' : 'cancellation_deadline_expired'); + if (! self::parallelCommandsMatchSequences($parsed['non_terminal'], $sequence, $run)) { + return $refused('invalid_' . $checkpointName . '_checkpoint_commands'); } - if (isset($cleanup['scope_id']) && now()->gte( - PortableLocalActivityCleanup::currentDeadline($run, $cleanup) - )) { - return $refused('local_activity_cleanup_authority_expired'); - } - } - $this->recordAppliedSignalForSignalResume($run, $task); - $this->recordSatisfiedConditionWaitForSignalResume($run, $task, $parsed['non_terminal']); - $createdTaskIds = []; - $localActivities = []; - foreach ($parsed['non_terminal'] as $command) { - if ($command['type'] === 'prepare_local_activity') { - $execution = PortableLocalActivityPreparation::admitGroupMember( + // Validate every local scope/cleanup member before creating any sibling. + foreach ($parsed['non_terminal'] as $offset => $command) { + if ($command['type'] !== 'prepare_local_activity') { + continue; + } + if (! CancellationScopeHistory::isRecordedBefore( $run, - $task, - $sequence, - [ - ...$command, - 'type' => 'record_local_activity', - ], - $checkpointId, - $fingerprint + $command['cancellation_scope_id'] ?? CancellationScopeHistory::ROOT_SCOPE_ID, + $sequence + $offset, + )) { + return $refused('local_activity_scope_not_recorded'); + } + $cleanup = PortableLocalActivityCleanup::snapshot( + $run, + $command['cancellation_cleanup'] ?? null, + $sequence + $offset, + $command['cancellation_scope_id'] ?? CancellationScopeHistory::ROOT_SCOPE_ID, ); - $localActivities[] = [ - 'sequence' => $sequence, - 'activity_execution_id' => $execution->id, - ]; - ++$sequence; - continue; + if ((isset($command['cancellation_cleanup']) && $cleanup === null) + || ($run->cancellation_request_command_id !== null + && ! PortableLocalActivityCleanup::belongsToRunRequest($run, $cleanup)) + || ($run->cancellation_request_command_id === null && isset($command['cancellation_cleanup']) && ! isset($cleanup['scope_id']))) { + return $refused('local_activity_cleanup_authority_mismatch'); + } + if ($cleanup !== null && now()->gte(PortableLocalActivityCleanup::snapshotDeadline($cleanup))) { + return $refused(isset($cleanup['scope_id']) ? 'local_activity_cleanup_deadline_expired' : 'cancellation_deadline_expired'); + } + if (isset($cleanup['scope_id']) && now()->gte( + PortableLocalActivityCleanup::currentDeadline($run, $cleanup) + )) { + return $refused('local_activity_cleanup_authority_expired'); + } } - $sequence = $this->applyNonTerminalCommand($run, $task, $command, $sequence, $createdTaskIds); - } - // Command application can update task payload (for example signal - // consumption). Preserve that state rather than the old snapshot. - $payload = is_array($task->payload) ? $task->payload : []; - $receipt = [ - 'checkpoint_id' => $checkpointId, - 'task_id' => $taskId, - 'workflow_run_id' => $run->id, - 'workflow_task_attempt' => $workflowTaskAttempt, - 'lease_owner' => $leaseOwner, - 'lease_expires_at' => $task->lease_expires_at->toISOString(), - 'start_sequence' => $startSequence, - 'next_sequence' => $sequence, - 'fingerprint' => $fingerprint, - 'created_task_ids' => $createdTaskIds, - 'recorded_at' => now() - ->toISOString(), - ...($group ? [ - 'local_activities' => $localActivities, - ] : []), - ]; - $task->forceFill([ - 'payload' => [ - ...$payload, - $receiptKey => $receipt, - ], - ])->save(); + $this->recordAppliedSignalForSignalResume($run, $task); + $this->recordSatisfiedConditionWaitForSignalResume($run, $task, $parsed['non_terminal']); + $createdTaskIds = []; + $localActivities = []; + foreach ($parsed['non_terminal'] as $command) { + if ($command['type'] === 'prepare_local_activity') { + $execution = PortableLocalActivityPreparation::admitGroupMember( + $run, + $task, + $sequence, + [ + ...$command, + 'type' => 'record_local_activity', + ], + $checkpointId, + $fingerprint + ); + $localActivities[] = [ + 'sequence' => $sequence, + 'activity_execution_id' => $execution->id, + ]; + ++$sequence; + continue; + } + $sequence = $this->applyNonTerminalCommand($run, $task, $command, $sequence, $createdTaskIds); + } + // Command application can update task payload (for example signal + // consumption). Preserve that state rather than the old snapshot. + $payload = is_array($task->payload) ? $task->payload : []; + $receipt = [ + 'checkpoint_id' => $checkpointId, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'workflow_task_attempt' => $workflowTaskAttempt, + 'lease_owner' => $leaseOwner, + 'lease_expires_at' => $task->lease_expires_at->toISOString(), + 'start_sequence' => $startSequence, + 'next_sequence' => $sequence, + 'fingerprint' => $fingerprint, + 'created_task_ids' => $createdTaskIds, + 'recorded_at' => now() + ->toISOString(), + ...($group ? [ + 'local_activities' => $localActivities, + ] : []), + ]; + $task->forceFill([ + 'payload' => [ + ...$payload, + $receiptKey => $receipt, + ], + ])->save(); - return [ - ...$receipt, - 'checkpointed' => true, - 'duplicate' => false, - 'reason' => null, - ]; - }, 5); + return [ + ...$receipt, + 'checkpointed' => true, + 'duplicate' => false, + 'reason' => null, + ]; + }, 5); + } catch (HistoryTransactionLimitExceededException $refusal) { + return $refused(self::meterRefusalReason($refusal)); + } } /** @@ -4466,6 +4550,33 @@ private function applyRecordLocalActivity( ? array_values($command['attempts']) : []; $now = now(); + // The normalizer bounds every timing field (#930); the total the walk + // below subtracts is also checked, and its earliest timestamp must be + // one every supported column type stores, before anything is written. + $backdatedMs = 0; + foreach ($attempts as $index => $attempt) { + $heartbeats = is_array($attempt['heartbeats'] ?? null) ? $attempt['heartbeats'] : []; + $lastHeartbeat = $heartbeats[array_key_last($heartbeats)] ?? null; + $span = max( + is_array($lastHeartbeat) && is_int( + $lastHeartbeat['elapsed_ms'] ?? null + ) ? $lastHeartbeat['elapsed_ms'] : 0, + is_int($attempt['duration_ms'] ?? null) ? $attempt['duration_ms'] : 0, + ); + $backdatedMs = HistoryTransactionMeter::checkedAdd($backdatedMs, $span); + if ($index < count($attempts) - 1 && is_int($attempt['backoff_seconds'] ?? null)) { + $backdatedMs = HistoryTransactionMeter::checkedAdd( + $backdatedMs, + HistoryTransactionMeter::checkedMultiply($attempt['backoff_seconds'], 1000), + ); + } + } + if ($now->getTimestampMs() - $backdatedMs < self::EARLIEST_STORED_TIMESTAMP_MS) { + throw HistoryTransactionLimitExceededException::refused( + HistoryTransactionLimitExceededException::TIMESTAMP_OUT_OF_RANGE, + 'The local activity report reaches back before the earliest storable timestamp.', + ); + } // Attempt reports carry elapsed durations rather than wall-clock // timestamps. Walk backward from completion so each attempt ends // before the following attempt's declared retry backoff. @@ -4507,6 +4618,7 @@ private function applyRecordLocalActivity( $lastAttempt = $attempts[array_key_last($attempts)]; /** @var ActivityExecution $execution */ + HistoryTransactionMeter::current()?->chargeRows(1); $execution = ActivityExecution::query()->create([ 'workflow_run_id' => $run->id, 'sequence' => $sequence, @@ -4570,6 +4682,7 @@ private function applyRecordLocalActivity( default => ActivityAttemptStatus::Failed, }; + HistoryTransactionMeter::current()?->chargeRows(1); $attemptRecord = ActivityAttempt::query()->create([ 'id' => $attemptId, 'workflow_run_id' => $run->id, @@ -4720,6 +4833,7 @@ private function applyRecordLocalActivity( : ($outcome === 'timed_out' ? ActivityTimeoutException::class : RuntimeException::class); + HistoryTransactionMeter::current()?->chargeRows(1); $failure = WorkflowFailure::query()->create([ 'workflow_run_id' => $run->id, 'source_kind' => 'activity_execution', diff --git a/src/V2/Support/ExternalPayloads.php b/src/V2/Support/ExternalPayloads.php index fe2d5789..e77da73f 100644 --- a/src/V2/Support/ExternalPayloads.php +++ b/src/V2/Support/ExternalPayloads.php @@ -9,6 +9,7 @@ use Workflow\Serializers\CodecRegistry; use Workflow\V2\Contracts\ExternalPayloadStorageDriver; use Workflow\V2\Contracts\ExternalPayloadStoragePolicy; +use Workflow\V2\Contracts\PredictableExternalPayloadStorageDriver; final class ExternalPayloads { @@ -48,7 +49,18 @@ public static function externalize( } $canonicalCodec = CodecRegistry::canonicalize($codec); - $reference = ExternalPayloadStorage::store($driver, $payload, $canonicalCodec); + // In a metered workflow task completion, a driver that predicts its + // URI has the object staged and stored only once the completion has + // passed every limit. Any other driver stores it now, as before, after + // its raw bytes are charged; a completion refused later leaves that + // object unreferenced. + $meter = HistoryTransactionMeter::current(); + if ($meter !== null && $driver instanceof PredictableExternalPayloadStorageDriver) { + $reference = $meter->stage($driver, $payload, $canonicalCodec); + } else { + $meter?->chargeUnstaged(strlen($payload)); + $reference = ExternalPayloadStorage::store($driver, $payload, $canonicalCodec); + } return self::encodeStoredEnvelope([ 'codec' => $canonicalCodec, diff --git a/src/V2/Support/HistoryTransactionMeter.php b/src/V2/Support/HistoryTransactionMeter.php new file mode 100644 index 00000000..410e943b --- /dev/null +++ b/src/V2/Support/HistoryTransactionMeter.php @@ -0,0 +1,348 @@ +|null The totals of the last attempt that committed its body. + */ + private static ?array $lastUsed = null; + + /** + * @var array + */ + private array $used = [ + self::EVENTS => 0, + self::BYTES => 0, + self::ROWS => 0, + self::STAGED_BYTES => 0, + ]; + + /** + * Staged objects keyed by driver instance and predicted URI. + * + * @var array + */ + private array $stage = []; + + private ?HistoryTransactionLimitExceededException $tripped = null; + + /** + * @param array $limits + */ + private function __construct( + private readonly array $limits + ) { + } + + /** + * Run one transaction attempt's body with a fresh meter, then store its + * staged payloads. Call it inside the database transaction callback, so + * a retried attempt starts over and a refusal rolls everything back. + * + * @template T + * @param callable(self): T $body + * @return T + */ + public static function run(callable $body): mixed + { + if (self::$current !== null) { + throw new LogicException('A history transaction meter is already running.'); + } + + $meter = new self([ + self::EVENTS => StructuralLimits::historyTransactionSizeLimit(), + self::BYTES => StructuralLimits::historyTransactionBytesLimit(), + self::ROWS => StructuralLimits::historyTransactionRowsLimit(), + self::STAGED_BYTES => StructuralLimits::externalPayloadTransactionBytesLimit(), + ]); + self::$current = $meter; + + try { + $result = $body($meter); + $meter->refuseIfTripped(); + $meter->flush(); + $meter->refuseIfTripped(); + self::$lastUsed = $meter->used; + + return $result; + } finally { + self::$current = null; + } + } + + public static function current(): ?self + { + return self::$current; + } + + /** + * Run server-owned work on stored state inside a metered completion + * without charging it: a run's deadline close, which no other path + * performs while the run's workflow task is open. + * + * @template T + * @param callable(): T $work + * @return T + */ + public static function unmetered(callable $work): mixed + { + $meter = self::$current; + self::$current = null; + + try { + return $work(); + } finally { + self::$current = $meter; + } + } + + /** + * The totals the last admitted attempt charged, for diagnostics and the + * row-accounting tests. + * + * @return array|null + */ + public static function lastUsed(): ?array + { + return self::$lastUsed; + } + + /** + * Overflow-checked addition for every budget aggregate. + */ + public static function checkedAdd(int $left, int $right): int + { + $sum = $left + $right; + if (! is_int($sum)) { + throw HistoryTransactionLimitExceededException::over( + 'history_transaction_arithmetic', + PHP_INT_MAX, + PHP_INT_MAX + ); + } + + return $sum; + } + + /** + * Overflow-checked multiplication for every budget aggregate. + */ + public static function checkedMultiply(int $left, int $right): int + { + $product = $left * $right; + if (! is_int($product)) { + throw HistoryTransactionLimitExceededException::over( + 'history_transaction_arithmetic', + PHP_INT_MAX, + PHP_INT_MAX + ); + } + + return $product; + } + + /** + * One history event about to be written, with its bytes measured as + * {@see HistoryBudget} measures a stored event. + * + * @param array $payload + */ + public function chargeEvent(string $eventType, array $payload): void + { + try { + $encoded = json_encode($payload, JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE); + } catch (JsonException) { + $this->trip(HistoryTransactionLimitExceededException::refused( + HistoryTransactionLimitExceededException::MEASURE_UNAVAILABLE, + 'The history event cannot be measured, so the completion is refused.', + )); + } + + $this->charge(self::EVENTS, 1); + $this->charge(self::BYTES, self::checkedAdd(strlen($eventType), strlen($encoded))); + $this->charge(self::ROWS, 1); + } + + /** + * Rows the next statement will insert, update, upsert or delete, + * including any cascade it causes. Charged before the statement runs. + */ + public function chargeRows(int $rows): void + { + if ($rows < 0) { + throw new LogicException('A write cannot affect a negative number of rows.'); + } + $this->charge(self::ROWS, $rows); + } + + /** + * How many more rows this completion may affect, or null when the row + * limit is disabled. Bulk write sites read at most this plus one row. + */ + public function remainingRows(): ?int + { + $limit = $this->limits[self::ROWS]; + + return $limit > 0 ? max(0, $limit - $this->used[self::ROWS]) : null; + } + + /** + * Stage an externalizable payload: the per-payload limit and the raw + * external payload byte limit are charged first (each occurrence counts), + * then the object is registered under its driver and predicted URI for + * {@see self::flush()}. The same driver predicting the same URI for other + * content, hash or codec fails closed. + */ + public function stage( + PredictableExternalPayloadStorageDriver $driver, + string $data, + string $codec + ): ExternalPayloadReference { + $size = strlen($data); + $this->chargePayload($size); + + $sha256 = hash('sha256', $data); + $uri = $driver->uriFor($sha256, $codec); + // The stage holds the driver, so its object id stays unique while staged. + $key = spl_object_id($driver) . "\0" . $uri; + $staged = $this->stage[$key] ?? null; + if ($staged === null) { + $this->stage[$key] = [ + 'driver' => $driver, + 'data' => $data, + 'sha256' => $sha256, + 'codec' => $codec, + 'uri' => $uri, + ]; + } elseif (! hash_equals($staged['sha256'], $sha256) + || $staged['codec'] !== $codec + || ! hash_equals($staged['data'], $data)) { + $this->trip(HistoryTransactionLimitExceededException::refused( + HistoryTransactionLimitExceededException::STAGED_URI_COLLISION, + 'The external payload driver predicted one URI for different payloads.', + )); + } + + return new ExternalPayloadReference(uri: $uri, sha256: $sha256, sizeBytes: $size, codec: $codec); + } + + /** + * Charge a payload that a driver without URI prediction stores during the + * completion: the per-payload and raw external payload byte limits apply + * before the driver's put, as for a staged payload. + */ + public function chargeUnstaged(int $size): void + { + $this->chargePayload($size); + } + + /** + * @return array + */ + public function used(): array + { + return $this->used; + } + + /** + * Store every staged object, in staging order. Each put must return the + * predicted URI; any failure rolls the completion back. An object already + * stored by this flush stays: content-addressed objects may be shared, so + * nothing is deleted, and reachability-based retention collects it. + */ + private function flush(): void + { + foreach ($this->stage as $entry) { + try { + $stored = $entry['driver']->put($entry['data'], $entry['sha256'], $entry['codec']); + } catch (Throwable $error) { + $this->stage = []; + + throw $error; + } + if ($stored !== $entry['uri']) { + $this->stage = []; + $this->trip(HistoryTransactionLimitExceededException::refused( + HistoryTransactionLimitExceededException::STAGED_URI_MISMATCH, + 'The external payload driver stored a payload under a different URI than it predicted.', + )); + } + } + $this->stage = []; + } + + private function chargePayload(int $size): void + { + $this->refuseIfTripped(); + $payloadLimit = StructuralLimits::payloadSizeLimit(); + if ($payloadLimit > 0 && $size > $payloadLimit) { + $this->trip(HistoryTransactionLimitExceededException::over(self::PAYLOAD_SIZE, $size, $payloadLimit)); + } + $this->charge(self::STAGED_BYTES, $size); + } + + private function charge(string $dimension, int $amount): void + { + $this->refuseIfTripped(); + $next = self::checkedAdd($this->used[$dimension], $amount); + $limit = $this->limits[$dimension]; + if ($limit > 0 && $next > $limit) { + $this->trip(HistoryTransactionLimitExceededException::over($dimension, $next, $limit)); + } + $this->used[$dimension] = $next; + } + + private function refuseIfTripped(): void + { + if ($this->tripped !== null) { + throw $this->tripped; + } + } + + private function trip(HistoryTransactionLimitExceededException $refusal): never + { + $this->tripped = $refusal; + $this->stage = []; + + throw $refusal; + } +} diff --git a/src/V2/Support/LifecycleEventDispatcher.php b/src/V2/Support/LifecycleEventDispatcher.php index 30c4d019..ffb2b2ee 100644 --- a/src/V2/Support/LifecycleEventDispatcher.php +++ b/src/V2/Support/LifecycleEventDispatcher.php @@ -4,6 +4,9 @@ namespace Workflow\V2\Support; +use Closure; +use Illuminate\Support\Facades\DB; +use Throwable; use Workflow\Events\ActivityCompleted as LegacyActivityCompleted; use Workflow\Events\ActivityFailed as LegacyActivityFailed; use Workflow\Events\ActivityStarted as LegacyActivityStarted; @@ -34,66 +37,174 @@ * Events carry scalar identity values captured eagerly from the models and * are dispatched synchronously within the same request/job that committed * the state change. + * + * A transaction body that may still roll back after dispatching runs under + * {@see self::collect()}: its events are held back and dispatched by + * {@see self::release()} once the transaction has committed. */ final class LifecycleEventDispatcher { + /** + * The events held back by each running transaction attempt, innermost last. + * + * @var list> + */ + private static array $collectors = []; + + /** + * Run one transaction attempt's body and hold back the lifecycle events it + * dispatches. Call it inside the DB::transaction() callback and pass the + * returned events to {@see self::release()} after DB::transaction() + * returns. Each call collects for its own attempt: if the body throws, + * its events are discarded, so a refused, failed or retried attempt + * dispatches nothing, even when an enclosing attempt catches the failure + * and commits. + * + * @template T + * @param callable(): T $body + * @return array{0: T, 1: list} + */ + public static function collect(callable $body): array + { + self::$collectors[] = []; + $depth = count(self::$collectors); + + try { + $result = $body(); + } catch (Throwable $error) { + array_splice(self::$collectors, $depth - 1); + + throw $error; + } + + $events = self::$collectors[$depth - 1]; + array_splice(self::$collectors, $depth - 1); + + return [$result, $events]; + } + + /** + * Release the events of an attempt whose transaction has committed. Inside + * an enclosing collecting attempt they join that attempt, and are + * dispatched or discarded with it. Otherwise, inside a caller's still-open + * transaction they wait for its commit, and outside any transaction they + * are dispatched now. + * + * @param list $events + */ + public static function release(array $events): void + { + if ($events === []) { + return; + } + + if (self::$collectors !== []) { + array_push(self::$collectors[count(self::$collectors) - 1], ...$events); + + return; + } + + $dispatch = static function () use ($events): void { + foreach ($events as $event) { + $event(); + } + }; + + if (DB::transactionLevel() > 0) { + DB::afterCommit($dispatch); + + return; + } + + $dispatch(); + } + public static function workflowStarted(WorkflowRun $run): void { $instanceId = (string) $run->instance?->id; + $runId = (string) $run->id; + $workflowType = (string) ($run->workflow_type ?? $run->workflow_class); + $workflowClass = (string) $run->workflow_class; $committedAt = now() ->toIso8601String(); + $stateChanged = self::stateChanged($run, null, new WorkflowRunningStatus($run)); - WorkflowStarted::dispatch( + self::emit(static function () use ( $instanceId, - (string) $run->id, - (string) ($run->workflow_type ?? $run->workflow_class), - (string) $run->workflow_class, + $runId, + $workflowType, + $workflowClass, $committedAt, - ); + $stateChanged + ): void { + WorkflowStarted::dispatch($instanceId, $runId, $workflowType, $workflowClass, $committedAt); - LegacyWorkflowStarted::dispatch($instanceId, (string) $run->workflow_class, '[]', $committedAt); + LegacyWorkflowStarted::dispatch($instanceId, $workflowClass, '[]', $committedAt); - self::dispatchStateChanged($run, null, new WorkflowRunningStatus($run)); + event($stateChanged); + }); } public static function workflowCompleted(WorkflowRun $run): void { $instanceId = (string) $run->instance?->id; + $runId = (string) $run->id; + $workflowType = (string) ($run->workflow_type ?? $run->workflow_class); + $workflowClass = (string) $run->workflow_class; $committedAt = now() ->toIso8601String(); + $stateChanged = self::stateChanged($run, new WorkflowRunningStatus($run), new WorkflowCompletedStatus($run)); - WorkflowCompleted::dispatch( + self::emit(static function () use ( $instanceId, - (string) $run->id, - (string) ($run->workflow_type ?? $run->workflow_class), - (string) $run->workflow_class, + $runId, + $workflowType, + $workflowClass, $committedAt, - ); + $stateChanged + ): void { + WorkflowCompleted::dispatch($instanceId, $runId, $workflowType, $workflowClass, $committedAt); - LegacyWorkflowCompleted::dispatch($instanceId, '', $committedAt); + LegacyWorkflowCompleted::dispatch($instanceId, '', $committedAt); - self::dispatchStateChanged($run, new WorkflowRunningStatus($run), new WorkflowCompletedStatus($run)); + event($stateChanged); + }); } public static function workflowFailed(WorkflowRun $run, string $exceptionClass, string $message): void { $instanceId = (string) $run->instance?->id; + $runId = (string) $run->id; + $workflowType = (string) ($run->workflow_type ?? $run->workflow_class); + $workflowClass = (string) $run->workflow_class; $committedAt = now() ->toIso8601String(); + $stateChanged = self::stateChanged($run, new WorkflowRunningStatus($run), new WorkflowFailedStatus($run)); - WorkflowFailed::dispatch( + self::emit(static function () use ( $instanceId, - (string) $run->id, - (string) ($run->workflow_type ?? $run->workflow_class), - (string) $run->workflow_class, + $runId, + $workflowType, + $workflowClass, $exceptionClass, $message, $committedAt, - ); + $stateChanged + ): void { + WorkflowFailed::dispatch( + $instanceId, + $runId, + $workflowType, + $workflowClass, + $exceptionClass, + $message, + $committedAt, + ); - LegacyWorkflowFailed::dispatch($instanceId, $exceptionClass . ': ' . $message, $committedAt); + LegacyWorkflowFailed::dispatch($instanceId, $exceptionClass . ': ' . $message, $committedAt); - self::dispatchStateChanged($run, new WorkflowRunningStatus($run), new WorkflowFailedStatus($run)); + event($stateChanged); + }); } public static function activityStarted( @@ -105,28 +216,40 @@ public static function activityStarted( int $attemptNumber, ): void { $instanceId = (string) $run->instance?->id; + $runId = (string) $run->id; $committedAt = now() ->toIso8601String(); - ActivityStarted::dispatch( + self::emit(static function () use ( $instanceId, - (string) $run->id, + $runId, $activityExecutionId, $activityType, $activityClass, $sequence, $attemptNumber, - $committedAt, - ); + $committedAt + ): void { + ActivityStarted::dispatch( + $instanceId, + $runId, + $activityExecutionId, + $activityType, + $activityClass, + $sequence, + $attemptNumber, + $committedAt, + ); - LegacyActivityStarted::dispatch( - $instanceId, - $activityExecutionId, - $activityClass, - $sequence, - '[]', - $committedAt, - ); + LegacyActivityStarted::dispatch( + $instanceId, + $activityExecutionId, + $activityClass, + $sequence, + '[]', + $committedAt, + ); + }); } public static function activityCompleted( @@ -138,28 +261,40 @@ public static function activityCompleted( int $attemptNumber, ): void { $instanceId = (string) $run->instance?->id; + $runId = (string) $run->id; $committedAt = now() ->toIso8601String(); - ActivityCompleted::dispatch( + self::emit(static function () use ( $instanceId, - (string) $run->id, + $runId, $activityExecutionId, $activityType, $activityClass, $sequence, $attemptNumber, - $committedAt, - ); + $committedAt + ): void { + ActivityCompleted::dispatch( + $instanceId, + $runId, + $activityExecutionId, + $activityType, + $activityClass, + $sequence, + $attemptNumber, + $committedAt, + ); - LegacyActivityCompleted::dispatch( - $instanceId, - $activityExecutionId, - '', - $committedAt, - $activityClass, - $sequence, - ); + LegacyActivityCompleted::dispatch( + $instanceId, + $activityExecutionId, + '', + $committedAt, + $activityClass, + $sequence, + ); + }); } public static function activityFailed( @@ -173,12 +308,13 @@ public static function activityFailed( string $message, ): void { $instanceId = (string) $run->instance?->id; + $runId = (string) $run->id; $committedAt = now() ->toIso8601String(); - ActivityFailed::dispatch( + self::emit(static function () use ( $instanceId, - (string) $run->id, + $runId, $activityExecutionId, $activityType, $activityClass, @@ -186,17 +322,30 @@ public static function activityFailed( $attemptNumber, $exceptionClass, $message, - $committedAt, - ); + $committedAt + ): void { + ActivityFailed::dispatch( + $instanceId, + $runId, + $activityExecutionId, + $activityType, + $activityClass, + $sequence, + $attemptNumber, + $exceptionClass, + $message, + $committedAt, + ); - LegacyActivityFailed::dispatch( - $instanceId, - $activityExecutionId, - $exceptionClass . ': ' . $message, - $committedAt, - $activityClass, - $sequence, - ); + LegacyActivityFailed::dispatch( + $instanceId, + $activityExecutionId, + $exceptionClass . ': ' . $message, + $committedAt, + $activityClass, + $sequence, + ); + }); } public static function failureRecorded( @@ -207,31 +356,62 @@ public static function failureRecorded( string $exceptionClass, string $message, ): void { - FailureRecorded::dispatch( - (string) $run->instance?->id, - (string) $run->id, + $instanceId = (string) $run->instance?->id; + $runId = (string) $run->id; + $committedAt = now() + ->toIso8601String(); + + self::emit(static function () use ( + $instanceId, + $runId, $failureId, $sourceKind, $sourceId, $exceptionClass, $message, - now() - ->toIso8601String(), - ); + $committedAt + ): void { + FailureRecorded::dispatch( + $instanceId, + $runId, + $failureId, + $sourceKind, + $sourceId, + $exceptionClass, + $message, + $committedAt, + ); + }); + } + + /** + * Dispatch now, or hold back for the collecting transaction attempt. + * + * @param Closure(): void $dispatch + */ + private static function emit(Closure $dispatch): void + { + if (self::$collectors !== []) { + self::$collectors[count(self::$collectors) - 1][] = $dispatch; + + return; + } + + $dispatch(); } /** - * Dispatch a V1-compatible StateChanged event over a run-status transition. + * A V1-compatible StateChanged event over a run-status transition. * * This is a compatibility adapter — V2 does not use V1 state machines, but * apps listening for StateChanged continue to receive notifications when * workflow status transitions occur. */ - private static function dispatchStateChanged( + private static function stateChanged( WorkflowRun $run, ?\Workflow\States\WorkflowStatus $initialState, \Workflow\States\WorkflowStatus $finalState, - ): void { - event(new StateChanged($initialState, $finalState, $run, 'status')); + ): StateChanged { + return new StateChanged($initialState, $finalState, $run, 'status'); } } diff --git a/src/V2/Support/LocalFilesystemExternalPayloadStorage.php b/src/V2/Support/LocalFilesystemExternalPayloadStorage.php index 55a32e81..9e8bf78a 100644 --- a/src/V2/Support/LocalFilesystemExternalPayloadStorage.php +++ b/src/V2/Support/LocalFilesystemExternalPayloadStorage.php @@ -4,17 +4,31 @@ namespace Workflow\V2\Support; +use Closure; use InvalidArgumentException; use RuntimeException; -use Workflow\V2\Contracts\ExternalPayloadStorageDriver; +use Workflow\V2\Contracts\PredictableExternalPayloadStorageDriver; use Workflow\V2\Exceptions\ExternalPayloadIntegrityException; -final class LocalFilesystemExternalPayloadStorage implements ExternalPayloadStorageDriver +final class LocalFilesystemExternalPayloadStorage implements PredictableExternalPayloadStorageDriver { private string $root; - public function __construct(string $root) + /** + * @var Closure(string, string): (int|false) + */ + private Closure $writeFile; + + /** + * @param (Closure(string, string): (int|false))|null $writeFile Writes a temporary object file and returns + * the bytes written; defaults to file_put_contents(). Tests replace it to prove that an incomplete or + * corrupt write is never moved into place. + */ + public function __construct(string $root, ?Closure $writeFile = null) { + $this->writeFile = $writeFile + ?? static fn (string $path, string $data): int|false => file_put_contents($path, $data, LOCK_EX); + $resolved = realpath($root); if ($resolved === false) { @@ -35,28 +49,59 @@ public function __construct(string $root) $this->root = rtrim($resolved, DIRECTORY_SEPARATOR); } + /** + * Store the content under its hash. The content must match the hash. An + * existing object is kept only if it holds exactly this content; a new, + * partial or corrupt one is replaced by a copy written aside, checked for + * its exact length and hash, and renamed into place. The object therefore + * appears only whole, and a repeated put leaves the same object. + */ public function put(string $data, string $sha256, string $codec): string { $this->validateSha256($sha256); - $codecSegment = $this->safeCodecSegment($codec); - $path = $this->root . DIRECTORY_SEPARATOR . $codecSegment . DIRECTORY_SEPARATOR . substr( - $sha256, - 0, - 2 - ) . DIRECTORY_SEPARATOR . $sha256; + if (! hash_equals(strtolower($sha256), hash('sha256', $data))) { + throw new ExternalPayloadIntegrityException('External payload content does not match its sha256.'); + } + $path = $this->pathFor($sha256, $codec); $directory = dirname($path); if (! is_dir($directory) && ! mkdir($directory, 0775, true) && ! is_dir($directory)) { throw new RuntimeException(sprintf('Unable to create external payload directory [%s].', $directory)); } - if (! is_file($path) && file_put_contents($path, $data, LOCK_EX) === false) { - throw new RuntimeException(sprintf('Unable to write external payload [%s].', $path)); + if ($this->holds($path, $sha256, strlen($data))) { + return self::pathToFileUri($path); + } + + $temporary = $path . '.' . bin2hex(random_bytes(8)) . '.tmp'; + + try { + $written = ($this->writeFile)($temporary, $data); + if ($written !== strlen($data) || ! $this->holds($temporary, $sha256, strlen($data))) { + throw new ExternalPayloadIntegrityException(sprintf( + 'External payload [%s] was not written completely.', + $path + )); + } + if (! rename($temporary, $path)) { + throw new RuntimeException(sprintf('Unable to write external payload [%s].', $path)); + } + } finally { + if (is_file($temporary)) { + unlink($temporary); + } } return self::pathToFileUri($path); } + public function uriFor(string $sha256, string $codec): string + { + $this->validateSha256($sha256); + + return self::pathToFileUri($this->pathFor($sha256, $codec)); + } + public function get(string $uri): string { $path = $this->pathFromUri($uri); @@ -82,6 +127,26 @@ public function delete(string $uri): void } } + /** + * Whether the file holds exactly the content with this length and hash. + */ + private function holds(string $path, string $sha256, int $size): bool + { + clearstatcache(true, $path); + if (! is_file($path) || filesize($path) !== $size) { + return false; + } + $actual = hash_file('sha256', $path); + + return $actual !== false && hash_equals(strtolower($sha256), $actual); + } + + private function pathFor(string $sha256, string $codec): string + { + return $this->root . DIRECTORY_SEPARATOR . $this->safeCodecSegment($codec) . DIRECTORY_SEPARATOR + . substr($sha256, 0, 2) . DIRECTORY_SEPARATOR . $sha256; + } + private function pathFromUri(string $uri): string { $parts = parse_url($uri); diff --git a/src/V2/Support/StructuralLimits.php b/src/V2/Support/StructuralLimits.php index 993e461b..43d6aac6 100644 --- a/src/V2/Support/StructuralLimits.php +++ b/src/V2/Support/StructuralLimits.php @@ -63,6 +63,12 @@ final class StructuralLimits public const DEFAULT_HISTORY_TRANSACTION_SIZE = 5000; + public const DEFAULT_HISTORY_TRANSACTION_BYTES = 16777216; // 16 MiB + + public const DEFAULT_HISTORY_TRANSACTION_ROWS = 20000; + + public const DEFAULT_EXTERNAL_PAYLOAD_TRANSACTION_BYTES = 16777216; // 16 MiB + public const DEFAULT_WARNING_THRESHOLD_PERCENT = 80; // ── Config readers ────────────────────────────────────────────── @@ -117,6 +123,39 @@ public static function historyTransactionSizeLimit(): int return self::intConfig('history_transaction_size', self::DEFAULT_HISTORY_TRANSACTION_SIZE); } + /** + * History bytes one workflow task completion may write, measured as + * {@see HistoryBudget} measures an event. 0 disables the check. + */ + public static function historyTransactionBytesLimit(): int + { + return self::intConfig('history_transaction_bytes', self::DEFAULT_HISTORY_TRANSACTION_BYTES); + } + + /** + * Metered rows one workflow task completion may write: each history event + * and the run position it advances, and each local activity execution, + * attempt and failure row. Other command writes are not counted yet. 0 + * disables the check. + */ + public static function historyTransactionRowsLimit(): int + { + return self::intConfig('history_transaction_rows', self::DEFAULT_HISTORY_TRANSACTION_ROWS); + } + + /** + * Raw payload bytes one workflow task completion may externalize, each + * occurrence counted even when objects deduplicate. 0 disables the + * check; the per-payload size limit still applies. + */ + public static function externalPayloadTransactionBytesLimit(): int + { + return self::intConfig( + 'external_payload_transaction_bytes', + self::DEFAULT_EXTERNAL_PAYLOAD_TRANSACTION_BYTES, + ); + } + public static function warningThresholdPercent(): int { return max(0, min(100, self::intConfig('warning_threshold_percent', self::DEFAULT_WARNING_THRESHOLD_PERCENT))); @@ -493,6 +532,9 @@ public static function limitForKind(StructuralLimitKind $kind): int StructuralLimitKind::MemoSize => self::memoSizeLimit(), StructuralLimitKind::SearchAttributeSize => self::searchAttributeSizeLimit(), StructuralLimitKind::HistoryTransactionSize => self::historyTransactionSizeLimit(), + StructuralLimitKind::HistoryTransactionBytes => self::historyTransactionBytesLimit(), + StructuralLimitKind::HistoryTransactionRows => self::historyTransactionRowsLimit(), + StructuralLimitKind::ExternalPayloadTransactionBytes => self::externalPayloadTransactionBytesLimit(), }; } @@ -515,6 +557,9 @@ public static function snapshot(): array 'memo_size_bytes' => self::memoSizeLimit(), 'search_attribute_size_bytes' => self::searchAttributeSizeLimit(), 'history_transaction_size' => self::historyTransactionSizeLimit(), + 'history_transaction_bytes' => self::historyTransactionBytesLimit(), + 'history_transaction_rows' => self::historyTransactionRowsLimit(), + 'external_payload_transaction_bytes' => self::externalPayloadTransactionBytesLimit(), 'warning_threshold_percent' => self::warningThresholdPercent(), ]; } diff --git a/src/V2/Support/WorkflowCommandNormalizer.php b/src/V2/Support/WorkflowCommandNormalizer.php index 218cdc6e..24e7b343 100644 --- a/src/V2/Support/WorkflowCommandNormalizer.php +++ b/src/V2/Support/WorkflowCommandNormalizer.php @@ -51,6 +51,30 @@ final class WorkflowCommandNormalizer public const MAX_LOCAL_ACTIVITY_HEARTBEATS = 1000; + /** + * The longest attempt duration_ms or heartbeat elapsed_ms a local + * activity report may declare: seven days. Recording walks each attempt + * back from completion by its duration and the following backoff, so + * with at most {@see self::MAX_LOCAL_ACTIVITY_ATTEMPTS} attempts every + * derived timestamp stays within a few years of now, which every + * supported timestamp type can hold. + */ + public const MAX_LOCAL_ACTIVITY_DURATION_MS = 604_800_000; + + /** + * The longest retry backoff_seconds a local activity attempt report may + * declare: one day. + */ + public const MAX_LOCAL_ACTIVITY_BACKOFF_SECONDS = 86_400; + + /** + * The history events the early completion check counts for each command + * other than record_local_activity: one, a lower bound. Several commands + * and every terminal path record more, depending on stored state; the + * completion's history transaction meter charges those as written. + */ + public const MIN_COMMAND_HISTORY_EVENTS = 1; + /** * @var list */ @@ -1267,6 +1291,86 @@ public static function normalize(array $commands, ?string $protocolVersion = nul return $normalized; } + /** + * The history events a normalized record_local_activity command records: + * one scheduled event; per attempt a started event, its heartbeats and, + * before the next attempt, a retry event; and one terminal event. + * + * @param array $command + */ + public static function localActivityHistoryEvents(array $command): int + { + $attempts = is_array($command['attempts'] ?? null) ? array_values($command['attempts']) : []; + $events = 2 + max(0, count($attempts) - 1); + foreach ($attempts as $attempt) { + $heartbeats = is_array($attempt) && is_array($attempt['heartbeats'] ?? null) ? $attempt['heartbeats'] : []; + $events += 1 + count($heartbeats); + } + + return $events; + } + + /** + * The exact history-event preflight for a whole completion, before any + * write (#930). Each local activity counts its own events (see + * {@see self::localActivityHistoryEvents()}) and every other command at + * least {@see self::MIN_COMMAND_HISTORY_EVENTS}. This refuses the + * local-activity amplifier before any work; the transaction's history + * meter remains the authority for every event, byte and row written. + * + * @param list> $normalized + * @return list the refusals, empty when the completion fits + */ + public static function completionBudgetErrors(array $normalized): array + { + $eventLimit = StructuralLimits::historyTransactionSizeLimit(); + $rowLimit = StructuralLimits::historyTransactionRowsLimit(); + $events = 0; + $rows = 0; + foreach ($normalized as $command) { + $local = is_array($command) && ($command['type'] ?? null) === 'record_local_activity'; + $count = $local ? self::localActivityHistoryEvents($command) : self::MIN_COMMAND_HISTORY_EVENTS; + $events = HistoryTransactionMeter::checkedAdd($events, $count); + $rows = HistoryTransactionMeter::checkedAdd($rows, HistoryTransactionMeter::checkedMultiply($count, 2)); + if ($local) { + $rows = HistoryTransactionMeter::checkedAdd($rows, self::localActivityRows($command)); + } + } + + $errors = []; + if ($eventLimit > 0 && $events > $eventLimit) { + $errors[] = sprintf( + 'This completion would record more than %d history events; one workflow task completion may record at most %d.', + $eventLimit, + $eventLimit, + ); + } + if ($rowLimit > 0 && $rows > $rowLimit) { + $errors[] = sprintf( + 'This completion would write more than %d durable rows; one workflow task completion may write at most %d.', + $rowLimit, + $rowLimit, + ); + } + + return $errors; + } + + /** + * The rows a normalized record_local_activity command writes besides its + * history (each event also writes the run's history position): the + * execution, each attempt, and a failure when it failed or timed out. + * + * @param array $command + */ + public static function localActivityRows(array $command): int + { + $attempts = is_array($command['attempts'] ?? null) ? count($command['attempts']) : 0; + $failure = in_array($command['outcome'] ?? null, ['failed', 'timed_out'], true) ? 1 : 0; + + return 1 + $attempts + $failure; + } + /** * @param array $command * @param array|null $retryPolicy @@ -1428,6 +1532,13 @@ private static function normalizeLocalActivityAttempts( $errors["{$path}.duration_ms"] = [ 'Local activity attempt duration_ms must be a non-negative integer.', ]; + } elseif ($rawAttempt['duration_ms'] > self::MAX_LOCAL_ACTIVITY_DURATION_MS) { + $errors["{$path}.duration_ms"] = [ + sprintf( + 'Local activity attempt duration_ms must not exceed %d.', + self::MAX_LOCAL_ACTIVITY_DURATION_MS, + ), + ]; } else { $durationMs = $rawAttempt['duration_ms']; } @@ -1487,6 +1598,13 @@ private static function normalizeLocalActivityAttempts( $errors["{$path}.backoff_seconds"] = [ 'Local activity attempt backoff_seconds must be a non-negative integer.', ]; + } elseif ($rawAttempt['backoff_seconds'] > self::MAX_LOCAL_ACTIVITY_BACKOFF_SECONDS) { + $errors["{$path}.backoff_seconds"] = [ + sprintf( + 'Local activity attempt backoff_seconds must not exceed %d.', + self::MAX_LOCAL_ACTIVITY_BACKOFF_SECONDS, + ), + ]; } else { $backoffSeconds = $rawAttempt['backoff_seconds']; } @@ -1504,11 +1622,15 @@ private static function normalizeLocalActivityAttempts( ), ]; } - if ($backoffSeconds === null) { + // A given value keeps its own refusal. + if (! array_key_exists('backoff_seconds', $rawAttempt)) { $errors["{$path}.backoff_seconds"] = [ 'A retried local activity attempt requires backoff_seconds.', ]; - } elseif (isset($backoffSchedule[$position]) && $backoffSchedule[$position] !== $backoffSeconds) { + } elseif ($backoffSeconds !== null + && isset($backoffSchedule[$position]) + && $backoffSchedule[$position] !== $backoffSeconds + ) { $errors["{$path}.backoff_seconds"] = [ 'Local activity attempt backoff_seconds must match retry_policy.backoff_seconds.', ]; @@ -1699,6 +1821,16 @@ private static function normalizeLocalActivityHeartbeats( continue; } + if ($elapsedMs > self::MAX_LOCAL_ACTIVITY_DURATION_MS) { + $errors["{$path}.elapsed_ms"] = [ + sprintf( + 'Local activity heartbeat elapsed_ms must not exceed %d.', + self::MAX_LOCAL_ACTIVITY_DURATION_MS, + ), + ]; + + continue; + } if ($elapsedMs < $lastElapsedMs) { $errors["{$path}.elapsed_ms"] = [ 'Local activity heartbeat elapsed_ms values must be ordered monotonically.', diff --git a/src/config/workflows.php b/src/config/workflows.php index 70f6c8b6..946c0d54 100644 --- a/src/config/workflows.php +++ b/src/config/workflows.php @@ -275,6 +275,25 @@ 'WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE', 5000 ), + // Hard ceilings for one workflow task completion; 0 disables a + // dimension. history_transaction_rows counts the rows history + // events and local activity records write; other command writes + // are not counted yet. + 'history_transaction_bytes' => (int) Env::dw( + 'DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES', + 'WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES', + 16777216 + ), + 'history_transaction_rows' => (int) Env::dw( + 'DW_V2_LIMIT_HISTORY_TRANSACTION_ROWS', + 'WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_ROWS', + 20000 + ), + 'external_payload_transaction_bytes' => (int) Env::dw( + 'DW_V2_LIMIT_EXTERNAL_PAYLOAD_TRANSACTION_BYTES', + 'WORKFLOW_V2_LIMIT_EXTERNAL_PAYLOAD_TRANSACTION_BYTES', + 16777216 + ), 'warning_threshold_percent' => (int) Env::dw( 'DW_V2_LIMIT_WARNING_THRESHOLD_PERCENT', 'WORKFLOW_V2_LIMIT_WARNING_THRESHOLD_PERCENT', diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index b2e78840..05ed98a9 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -10,9 +10,12 @@ use Illuminate\Support\Facades\Artisan; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; +use Illuminate\Validation\ValidationException; use PHPUnit\Framework\Attributes\PreserveGlobalState; use PHPUnit\Framework\Attributes\RunInSeparateProcess; use Symfony\Component\Uid\Ulid; +use Tests\Fixtures\V2\TestGreetingActivity; +use Tests\Fixtures\V2\TestLocalActivityWorkflow; use Tests\Fixtures\V2\TestQueryReplayGuardActivity; use Tests\Fixtures\V2\TestRepeatedPatchClockWorkflow; use Tests\Fixtures\V2\TestSinglePatchReplayWorkflow; @@ -417,6 +420,10 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void $this->assertPortableLocalActivityAttemptTimelineReplay($fixture); } + if (($fixture['id'] ?? null) === 'portable-local-activity-timing-caps-cold-replay') { + $this->assertLocalActivityTimingCapsBoundTheRecordedWalk($fixture); + } + if (($fixture['id'] ?? null) === 'portable-local-activity-failure-avro-cold-replay') { $this->assertPortableLocalActivityFailureUsesAvro($fixture); } @@ -1471,6 +1478,136 @@ private function assertPortableLocalActivityAttemptTimelineReplay(array $fixture ->all()); } + /** + * The longest local activity timings a worker may report record exactly + * the attempt walk this fixture replays, and one past any of them is + * refused before anything is written. + * + * @param array $fixture + */ + private function assertLocalActivityTimingCapsBoundTheRecordedWalk(array $fixture): void + { + $report = $fixture['local_activity_report']; + $expected = []; + foreach ($fixture['history'] as $event) { + $attempt = $event['payload']['activity_attempt'] ?? null; + if (is_array($attempt)) { + $expected[$attempt['attempt_number']] = [ + $attempt['started_at'] ?? null, + $attempt['last_heartbeat_at'] ?? null, + $attempt['closed_at'] ?? null, + ]; + } + } + ksort($expected); + $command = static fn (array $attempts): array => [ + 'type' => 'record_local_activity', + 'activity_type' => TestGreetingActivity::class, + 'arguments' => Serializer::serializeWithCodec('avro', $fixture['workflow']['arguments']), + 'result' => Serializer::serializeWithCodec('avro', 'Hello, Ada!'), + 'payload_codec' => 'avro', + 'outcome' => 'completed', + 'attempts' => $attempts, + 'retry_policy' => [ + 'max_attempts' => count($attempts), + ], + 'execution_mode' => 'local', + ]; + $bridge = $this->app->make(WorkflowTaskBridge::class); + + Carbon::setTestNow($report['completed_at']); + + try { + [$run, $task] = $this->leasedLocalActivityRun(); + $accepted = $bridge->complete($task->id, [$command($report['attempts'])]); + $this->assertTrue($accepted['completed'] ?? false, json_encode($accepted, JSON_THROW_ON_ERROR)); + $recorded = []; + $attempts = ActivityAttempt::query() + ->where('workflow_run_id', $run->id) + ->orderBy('attempt_number') + ->get(); + foreach ($attempts as $attempt) { + $recorded[$attempt->attempt_number] = [ + $attempt->started_at?->toJSON(), + $attempt->last_heartbeat_at?->toJSON(), + $attempt->closed_at?->toJSON(), + ]; + } + $this->assertSame($expected, $recorded); + + foreach ($report['one_over'] as $case) { + $over = $command($report['attempts']); + data_set($over, $case['field'], $case['value']); + $errors = []; + try { + WorkflowCommandNormalizer::normalize([$over]); + } catch (ValidationException $refusal) { + $errors = $refusal->errors(); + } + $this->assertSame( + ["commands.0.{$case['field']}"], + array_keys($errors), + "{$case['field']} one past its cap must be refused by that field alone." + ); + + [$run, $task] = $this->leasedLocalActivityRun(); + $refused = $bridge->complete($task->id, [$over]); + + $this->assertFalse($refused['completed'] ?? true, "{$case['field']} one past its cap was accepted."); + $this->assertSame(0, ActivityAttempt::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(TaskStatus::Leased, $task->fresh()->status); + } + } finally { + Carbon::setTestNow(); + } + } + + /** + * @return array{WorkflowRun, WorkflowTask} + */ + private function leasedLocalActivityRun(): array + { + $instance = WorkflowInstance::query()->create([ + 'workflow_class' => TestLocalActivityWorkflow::class, + 'workflow_type' => TestLocalActivityWorkflow::class, + 'run_count' => 1, + 'started_at' => now() + ->subMinute(), + ]); + $run = WorkflowRun::query()->create([ + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'workflow_class' => TestLocalActivityWorkflow::class, + 'workflow_type' => TestLocalActivityWorkflow::class, + 'status' => RunStatus::Waiting, + 'arguments' => Serializer::serializeWithCodec('avro', ['Ada']), + 'payload_codec' => 'avro', + 'connection' => 'database', + 'queue' => 'default', + 'compatibility' => 'build-a', + 'started_at' => now() + ->subMinute(), + ]); + $instance->forceFill([ + 'current_run_id' => $run->id, + ])->save(); + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $run->id, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Leased, + 'attempt_count' => 1, + 'payload' => [], + 'connection' => 'database', + 'queue' => 'default', + 'compatibility' => 'build-a', + 'lease_owner' => 'timing-caps-worker', + 'lease_expires_at' => now() + ->addMinutes(5), + ]); + + return [$run, $task]; + } + /** * @param array $attemptsByActivity */ diff --git a/tests/Fixtures/V2/RecordingExternalPayloadStorage.php b/tests/Fixtures/V2/RecordingExternalPayloadStorage.php new file mode 100644 index 00000000..7b689db1 --- /dev/null +++ b/tests/Fixtures/V2/RecordingExternalPayloadStorage.php @@ -0,0 +1,117 @@ + + */ + public array $puts = []; + + /** + * @var array + */ + public array $objects = []; + + public bool $failPuts = false; + + private readonly ExternalPayloadStorageDriver $driver; + + public function __construct( + bool $predictable = true, + private readonly int $threshold = 16 + ) { + $recorder = $this; + $this->driver = $predictable + ? new class($recorder) implements PredictableExternalPayloadStorageDriver { + public function __construct( + private readonly RecordingExternalPayloadStorage $recorder + ) { + } + + public function uriFor(string $sha256, string $codec): string + { + return "memory://{$codec}/{$sha256}"; + } + + public function put(string $data, string $sha256, string $codec): string + { + if ($this->recorder->failPuts) { + throw new RuntimeException('put failed'); + } + + return $this->recorder->record($data, $sha256, $codec, $this->uriFor($sha256, $codec)); + } + + public function get(string $uri): string + { + return $this->recorder->objects[$uri] ?? throw new RuntimeException("No payload at {$uri}."); + } + + public function delete(string $uri): void + { + } + } + : new class($recorder) implements ExternalPayloadStorageDriver { + public function __construct( + private readonly RecordingExternalPayloadStorage $recorder + ) { + } + + public function put(string $data, string $sha256, string $codec): string + { + return $this->recorder->record( + $data, + $sha256, + $codec, + 'memory://random/' . bin2hex(random_bytes(8)) + ); + } + + public function get(string $uri): string + { + return $this->recorder->objects[$uri] ?? throw new RuntimeException("No payload at {$uri}."); + } + + public function delete(string $uri): void + { + } + }; + } + + public function driverFor(?string $namespace): ?ExternalPayloadStorageDriver + { + return $this->driver; + } + + public function thresholdBytesFor(?string $namespace): ?int + { + return $this->threshold; + } + + public function record(string $data, string $sha256, string $codec, string $uri): string + { + $this->puts[] = [ + 'sha256' => $sha256, + 'codec' => $codec, + ]; + $this->objects[$uri] = $data; + + return $uri; + } +} diff --git a/tests/Fixtures/V2/ReplayRegression/portable-local-activity-timing-caps-cold-replay.json b/tests/Fixtures/V2/ReplayRegression/portable-local-activity-timing-caps-cold-replay.json new file mode 100644 index 00000000..f8b35feb --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/portable-local-activity-timing-caps-cold-replay.json @@ -0,0 +1,355 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "portable-local-activity-timing-caps-cold-replay", + "protocol_version": "1.19", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner", + "embedded-history-import" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestLocalActivityWorkflow", + "arguments": [ + "Ada" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-08-13T11:59:52+00:00" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "payload": { + "activity_execution_id": "01j00000000000000000000001", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "execution_mode": "local", + "local_activity": true, + "activity": { + "id": "01j00000000000000000000001", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "execution_mode": "local", + "local_activity": true, + "status": "pending", + "payload_codec": "avro", + "attempt_count": 0 + } + }, + "recorded_at": "2026-08-28T12:00:00+00:00" + }, + { + "sequence": 3, + "event_type": "ActivityStarted", + "payload": { + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000a", + "worker_attempt_id": "worker-a-attempt-1", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "attempt_number": 1, + "execution_mode": "local", + "local_activity": true, + "activity": { + "id": "01j00000000000000000000001", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "execution_mode": "local", + "local_activity": true, + "attempt_id": "01j0000000000000000000000a", + "status": "running", + "payload_codec": "avro", + "attempt_count": 1, + "started_at": "2026-08-13T12:00:00.000000Z" + }, + "activity_attempt": { + "id": "01j0000000000000000000000a", + "worker_attempt_id": "worker-a-attempt-1", + "attempt_number": 1, + "status": "running", + "started_at": "2026-08-13T12:00:00.000000Z" + } + }, + "recorded_at": "2026-08-28T12:00:00+00:00" + }, + { + "sequence": 4, + "event_type": "ActivityHeartbeatRecorded", + "payload": { + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000a", + "worker_attempt_id": "worker-a-attempt-1", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "attempt_number": 1, + "heartbeat_at": "2026-08-13T12:00:01.500000Z", + "lease_expires_at": null, + "execution_mode": "local", + "local_activity": true, + "progress": { + "worker": "worker-a" + }, + "activity": { + "id": "01j00000000000000000000001", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "execution_mode": "local", + "local_activity": true, + "attempt_id": "01j0000000000000000000000a", + "status": "running", + "payload_codec": "avro", + "attempt_count": 1, + "started_at": "2026-08-13T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-13T12:00:01.500000Z" + }, + "activity_attempt": { + "id": "01j0000000000000000000000a", + "worker_attempt_id": "worker-a-attempt-1", + "attempt_number": 1, + "status": "running", + "started_at": "2026-08-13T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-13T12:00:01.500000Z" + } + }, + "recorded_at": "2026-08-28T12:00:00+00:00" + }, + { + "sequence": 5, + "event_type": "ActivityRetryScheduled", + "payload": { + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000a", + "worker_attempt_id": "worker-a-attempt-1", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "retry_after_attempt_id": "01j0000000000000000000000a", + "retry_after_attempt": 1, + "retry_reason": "failure", + "retry_backoff_seconds": 86400, + "message": "retry on another worker", + "execution_mode": "local", + "local_activity": true, + "activity": { + "id": "01j00000000000000000000001", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "execution_mode": "local", + "local_activity": true, + "attempt_id": "01j0000000000000000000000a", + "status": "pending", + "payload_codec": "avro", + "attempt_count": 1, + "started_at": "2026-08-13T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-13T12:00:01.500000Z" + }, + "activity_attempt": { + "id": "01j0000000000000000000000a", + "worker_attempt_id": "worker-a-attempt-1", + "attempt_number": 1, + "status": "failed", + "started_at": "2026-08-13T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-13T12:00:01.500000Z", + "closed_at": "2026-08-20T12:00:00.000000Z" + } + }, + "recorded_at": "2026-08-28T12:00:00+00:00" + }, + { + "sequence": 6, + "event_type": "ActivityStarted", + "payload": { + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000b", + "worker_attempt_id": "worker-b-attempt-1", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "attempt_number": 2, + "execution_mode": "local", + "local_activity": true, + "activity": { + "id": "01j00000000000000000000001", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "execution_mode": "local", + "local_activity": true, + "attempt_id": "01j0000000000000000000000b", + "status": "running", + "payload_codec": "avro", + "attempt_count": 2, + "started_at": "2026-08-13T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-13T12:00:01.500000Z" + }, + "activity_attempt": { + "id": "01j0000000000000000000000b", + "worker_attempt_id": "worker-b-attempt-1", + "attempt_number": 2, + "status": "running", + "started_at": "2026-08-21T12:00:00.000000Z" + } + }, + "recorded_at": "2026-08-28T12:00:00+00:00" + }, + { + "sequence": 7, + "event_type": "ActivityHeartbeatRecorded", + "payload": { + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000b", + "worker_attempt_id": "worker-b-attempt-1", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "attempt_number": 2, + "heartbeat_at": "2026-08-21T12:00:00.500000Z", + "lease_expires_at": null, + "execution_mode": "local", + "local_activity": true, + "progress": { + "worker": "worker-b" + }, + "activity": { + "id": "01j00000000000000000000001", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "execution_mode": "local", + "local_activity": true, + "attempt_id": "01j0000000000000000000000b", + "status": "running", + "payload_codec": "avro", + "attempt_count": 2, + "started_at": "2026-08-13T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-21T12:00:00.500000Z" + }, + "activity_attempt": { + "id": "01j0000000000000000000000b", + "worker_attempt_id": "worker-b-attempt-1", + "attempt_number": 2, + "status": "running", + "started_at": "2026-08-21T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-21T12:00:00.500000Z" + } + }, + "recorded_at": "2026-08-28T12:00:00+00:00" + }, + { + "sequence": 8, + "event_type": "ActivityCompleted", + "payload": { + "activity_execution_id": "01j00000000000000000000001", + "activity_attempt_id": "01j0000000000000000000000b", + "worker_attempt_id": "worker-b-attempt-1", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "attempt_number": 2, + "result": "wwHioz3/VYAiNwoWSGVsbG8sIEFkYSE=", + "payload_codec": "avro", + "execution_mode": "local", + "local_activity": true, + "activity": { + "id": "01j00000000000000000000001", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "execution_mode": "local", + "local_activity": true, + "attempt_id": "01j0000000000000000000000b", + "status": "completed", + "payload_codec": "avro", + "attempt_count": 2, + "started_at": "2026-08-13T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-21T12:00:00.500000Z", + "closed_at": "2026-08-28T12:00:00.000000Z" + }, + "activity_attempt": { + "id": "01j0000000000000000000000b", + "worker_attempt_id": "worker-b-attempt-1", + "attempt_number": 2, + "status": "completed", + "started_at": "2026-08-21T12:00:00.000000Z", + "last_heartbeat_at": "2026-08-21T12:00:00.500000Z", + "closed_at": "2026-08-28T12:00:00.000000Z" + } + }, + "recorded_at": "2026-08-28T12:00:00+00:00" + } + ], + "history_import_metadata": { + "memo": {}, + "search_attributes": {} + }, + "local_activity_report": { + "completed_at": "2026-08-28T12:00:00.000000Z", + "attempts": [ + { + "attempt_number": 1, + "outcome": "failed", + "duration_ms": 604800000, + "message": "retry on another worker", + "retry_reason": "failure", + "backoff_seconds": 86400, + "heartbeats": [ + { + "elapsed_ms": 1500 + } + ] + }, + { + "attempt_number": 2, + "outcome": "completed", + "duration_ms": 604800000, + "heartbeats": [ + { + "elapsed_ms": 500 + } + ] + } + ], + "one_over": [ + { + "field": "attempts.0.duration_ms", + "value": 604800001 + }, + { + "field": "attempts.1.heartbeats.0.elapsed_ms", + "value": 604800001 + }, + { + "field": "attempts.0.backoff_seconds", + "value": 86401 + } + ] + }, + "expected": { + "completed": true, + "result": { + "greeting": "Hello, Ada!", + "workflow_id": "regression-corpus-portable-local-activity-timing-caps-cold-replay", + "run_id": "regression-corpus-run-portable-local-activity-timing-caps-cold-replay" + }, + "commands": [ + { + "type": "complete_workflow" + } + ] + } +} diff --git a/tests/Unit/V2/DefaultWorkflowTaskBridgeLocalActivityTest.php b/tests/Unit/V2/DefaultWorkflowTaskBridgeLocalActivityTest.php index b8ea873a..db9c5cef 100644 --- a/tests/Unit/V2/DefaultWorkflowTaskBridgeLocalActivityTest.php +++ b/tests/Unit/V2/DefaultWorkflowTaskBridgeLocalActivityTest.php @@ -5,20 +5,33 @@ namespace Tests\Unit\V2; use Illuminate\Support\Carbon; +use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Event; +use RuntimeException; +use Tests\Fixtures\V2\RecordingExternalPayloadStorage; use Tests\Fixtures\V2\TestGreetingWorkflow; use Tests\TestCase; use Workflow\Serializers\Serializer; +use Workflow\V2\Contracts\ExternalPayloadStoragePolicy; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; +use Workflow\V2\Enums\TimerStatus; +use Workflow\V2\Events\ActivityFailed; +use Workflow\V2\Events\FailureRecorded; +use Workflow\V2\Events\WorkflowCompleted; use Workflow\V2\Models\ActivityAttempt; use Workflow\V2\Models\ActivityExecution; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Models\WorkflowTimer; +use Workflow\V2\Support\ExternalPayloads; +use Workflow\V2\Support\HistoryTransactionMeter; +use Workflow\V2\Support\WorkflowCommandNormalizer; final class DefaultWorkflowTaskBridgeLocalActivityTest extends TestCase { @@ -169,6 +182,513 @@ public function testRecordLocalActivityReconstructsAttemptTimelineAndLifecycleSn } } + /** + * Workflow #930: a completion whose history outgrows one transaction is + * refused before anything is written, and the worker keeps its lease. + */ + public function testAnOverBudgetCompletionWritesNothingAndKeepsTheLease(): void + { + config([ + 'workflows.v2.structural_limits.history_transaction_size' => 6, + ]); + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $command = static fn (int $heartbeats): array => [ + 'type' => 'record_local_activity', + 'activity_type' => 'bounded-local-activity', + 'result' => Serializer::serialize('done'), + 'outcome' => 'completed', + 'attempts' => [[ + 'attempt_number' => 1, + 'outcome' => 'completed', + 'heartbeats' => array_map( + static fn (int $elapsed): array => [ + 'elapsed_ms' => $elapsed, + ], + $heartbeats === 0 ? [] : range(1, $heartbeats) + ), + ]], + 'execution_mode' => 'local', + ]; + $before = [$task->fresh()->status, $task->fresh()->lease_owner, $task->fresh()->lease_expires_at?->toJSON(), + $run->fresh() +->status, $run->fresh() +->last_history_sequence]; + + // The local activity records six events (scheduled, started, three + // heartbeats, terminal) and the closing command at least one: seven. + $refused = $bridge->complete($task->id, [ + $command(3), [ + 'type' => 'complete_workflow', + 'result' => Serializer::serialize('x'), + ]]); + + $this->assertFalse($refused['completed']); + $this->assertSame('completion_budget_exceeded', $refused['reason']); + $this->assertSame(0, ActivityExecution::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(0, ActivityAttempt::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(0, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame($before, [$task->fresh()->status, $task->fresh()->lease_owner, + $task->fresh() + ->lease_expires_at?->toJSON(), $run->fresh() +->status, $run->fresh() +->last_history_sequence]); + + // The same lease completes a completion that fits. + $accepted = $bridge->complete($task->id, [$command(3)]); + $this->assertTrue($accepted['completed'], json_encode($accepted, JSON_THROW_ON_ERROR)); + $this->assertSame(6, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + } + + /** + * Workflow #930: the longest declared durations and backoffs, across the + * most attempts, still produce representable timestamps. + */ + public function testTheLongestDeclaredTimingsRecordRepresentableTimestamps(): void + { + Carbon::setTestNow('2026-08-28T12:00:00.000000Z'); + + try { + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $maximum = WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_DURATION_MS; + $backoff = WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_BACKOFF_SECONDS; + $count = WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_ATTEMPTS; + $attempts = []; + for ($number = 1; $number <= $count; $number++) { + $attempts[] = array_filter([ + 'attempt_number' => $number, + 'outcome' => $number === $count ? 'completed' : 'failed', + 'duration_ms' => $maximum, + 'message' => $number === $count ? null : 'retry', + 'retry_reason' => $number === $count ? null : 'failure', + 'backoff_seconds' => $number === $count ? null : $backoff, + 'heartbeats' => [[ + 'elapsed_ms' => $maximum, + ]], + ], static fn (mixed $value): bool => $value !== null); + } + $result = $bridge->complete($task->id, [[ + 'type' => 'record_local_activity', + 'activity_type' => 'longest-local-activity', + 'result' => Serializer::serialize('done'), + 'outcome' => 'completed', + 'attempts' => $attempts, + 'retry_policy' => [ + 'max_attempts' => $count, + ], + 'execution_mode' => 'local', + ]]); + + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $first = ActivityAttempt::query()->where('workflow_run_id', $run->id)->where('attempt_number', 1)->sole(); + // 100 seven-day attempts and 99 one-day backoffs before completion. + $this->assertSame( + Carbon::parse('2026-08-28T12:00:00.000000Z')->subMilliseconds($maximum * $count)->subSeconds( + $backoff * ($count - 1) + )->toJSON(), + $first->started_at?->toJSON(), + ); + } finally { + Carbon::setTestNow(); + } + } + + /** + * Workflow #930: past the history byte limit the meter refuses the whole + * completion; nothing is written and the same lease completes later. + */ + public function testTheMeterRefusesPastTheHistoryByteLimitAndKeepsTheLease(): void + { + config([ + 'workflows.v2.structural_limits.history_transaction_bytes' => 4000, + ]); + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $before = $this->leaseOf($task); + + $refused = $bridge->complete($task->id, [$this->localActivity(5)]); + + $this->assertFalse($refused['completed']); + $this->assertSame('history_transaction_limit_exceeded', $refused['reason']); + $this->assertSame('history_transaction_bytes', $refused['limit']); + $this->assertNothingRecorded($run); + $this->assertSame($before, $this->leaseOf($task)); + + config([ + 'workflows.v2.structural_limits.history_transaction_bytes' => 0, + ]); + $accepted = $bridge->complete($task->id, [$this->localActivity(5)]); + $this->assertTrue($accepted['completed'], json_encode($accepted, JSON_THROW_ON_ERROR)); + $this->assertSame(8, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + } + + /** + * Workflow #930: a local activity's externalized payloads are staged and + * stored only once the completion is admitted; a refused completion + * stores nothing. + */ + public function testExternalizedLocalActivityPayloadsAreStoredOnlyAfterAdmission(): void + { + $storage = new RecordingExternalPayloadStorage(); + $this->app->instance(ExternalPayloadStoragePolicy::class, $storage); + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $command = [ + ...$this->localActivity(0), + 'arguments' => Serializer::serialize([str_repeat('argument ', 20)]), + 'result' => Serializer::serialize(str_repeat('result ', 20)), + ]; + + config([ + // The meter refuses at the first event, after both payloads were staged. + 'workflows.v2.structural_limits.history_transaction_bytes' => 1000, + ]); + $refused = $bridge->complete($task->id, [$command]); + $this->assertSame('history_transaction_limit_exceeded', $refused['reason']); + $this->assertSame('history_transaction_bytes', $refused['limit']); + $this->assertSame([], $storage->puts); + $this->assertNothingRecorded($run); + + config([ + 'workflows.v2.structural_limits.history_transaction_bytes' => null, + ]); + $accepted = $bridge->complete($task->id, [$command]); + $this->assertTrue($accepted['completed'], json_encode($accepted, JSON_THROW_ON_ERROR)); + $this->assertSame( + [hash('sha256', $command['arguments']), hash('sha256', $command['result'])], + array_column($storage->puts, 'sha256') + ); + $execution = ActivityExecution::query()->where('workflow_run_id', $run->id)->sole(); + $envelope = ExternalPayloads::storedEnvelope((string) $execution->getRawOriginal('arguments')); + $this->assertSame( + 'memory://avro/' . hash('sha256', $command['arguments']), + $envelope['external_storage']['uri'] ?? null + ); + } + + /** + * A driver that implements only the original interface keeps working: it + * stores each payload when it is externalized, after the payload's raw + * bytes are charged, so a refused completion stores at most that limit. + */ + public function testADriverWithoutUriPredictionKeepsStoringWithinTheByteLimit(): void + { + $storage = new RecordingExternalPayloadStorage(predictable: false); + $this->app->instance(ExternalPayloadStoragePolicy::class, $storage); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $command = [ + ...$this->localActivity(0), + 'arguments' => Serializer::serialize([str_repeat('argument ', 20)]), + 'result' => Serializer::serialize(str_repeat('result ', 20)), + ]; + + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $accepted = $bridge->complete($task->id, [$command]); + $this->assertTrue($accepted['completed'], json_encode($accepted, JSON_THROW_ON_ERROR)); + $this->assertCount(2, $storage->puts); + $execution = ActivityExecution::query()->where('workflow_run_id', $run->id)->sole(); + $uri = ExternalPayloads::storedEnvelope( + (string) $execution->getRawOriginal('arguments') + )['external_storage']['uri'] ?? null; + $this->assertSame($command['arguments'], $storage->objects[$uri] ?? null); + + // Room for the arguments only: the result is refused before its put. + config([ + 'workflows.v2.structural_limits.external_payload_transaction_bytes' => strlen($command['arguments']), + ]); + $storage->puts = []; + $otherRun = $this->createWaitingRun(); + $otherTask = $this->createLeasedTask($otherRun); + $refused = $bridge->complete($otherTask->id, [$command]); + $this->assertSame('history_transaction_limit_exceeded', $refused['reason']); + $this->assertSame('external_payload_transaction_bytes', $refused['limit']); + $this->assertSame([hash('sha256', $command['arguments'])], array_column($storage->puts, 'sha256')); + $this->assertNothingRecorded($otherRun); + } + + /** + * Lifecycle events report committed state. A completion refused after a + * local activity failed dispatches none; once accepted, each event is + * dispatched exactly once. + */ + public function testLifecycleEventsAreDispatchedOnlyAfterTheCompletionCommits(): void + { + Event::fake([ActivityFailed::class, FailureRecorded::class, WorkflowCompleted::class]); + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $before = $this->leaseOf($task); + $commands = [$this->failedLocalActivity(), [ + 'type' => 'complete_workflow', + 'result' => Serializer::serialize(str_repeat('large result ', 20_000)), + ]]; + + // The local activity's events fit; the completion's large result does not. + config([ + 'workflows.v2.structural_limits.history_transaction_bytes' => 100_000, + ]); + $refused = $bridge->complete($task->id, $commands); + + $this->assertSame('history_transaction_limit_exceeded', $refused['reason']); + $this->assertSame('history_transaction_bytes', $refused['limit']); + Event::assertNotDispatched(ActivityFailed::class); + Event::assertNotDispatched(FailureRecorded::class); + Event::assertNotDispatched(WorkflowCompleted::class); + $this->assertNothingRecorded($run); + $this->assertSame($before, $this->leaseOf($task)); + + config([ + 'workflows.v2.structural_limits.history_transaction_bytes' => null, + ]); + $accepted = $bridge->complete($task->id, $commands); + + $this->assertTrue($accepted['completed'], json_encode($accepted, JSON_THROW_ON_ERROR)); + Event::assertDispatchedTimes(ActivityFailed::class, 1); + Event::assertDispatchedTimes(FailureRecorded::class, 1); + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + } + + public function testLifecycleListenersRunAfterTheCompletionTransactionCloses(): void + { + $levels = []; + Event::listen(WorkflowCompleted::class, static function () use (&$levels): void { + $levels[] = DB::transactionLevel(); + }); + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + + $accepted = $this->app->make(WorkflowTaskBridge::class)->complete($task->id, [[ + 'type' => 'complete_workflow', + 'result' => Serializer::serialize('done'), + ]]); + + $this->assertTrue($accepted['completed'], json_encode($accepted, JSON_THROW_ON_ERROR)); + $this->assertSame([0], $levels); + } + + /** + * A staged payload whose put fails rolls the completion back after its + * body dispatched lifecycle events; none of them is dispatched. + */ + public function testAFailedStagedPutDispatchesNoLifecycleEvent(): void + { + Event::fake([WorkflowCompleted::class]); + $storage = new RecordingExternalPayloadStorage(); + $storage->failPuts = true; + $this->app->instance(ExternalPayloadStoragePolicy::class, $storage); + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $before = $this->leaseOf($task); + $commands = [[ + 'type' => 'complete_workflow', + 'result' => Serializer::serialize(str_repeat('externalized result ', 4)), + ]]; + + try { + $bridge->complete($task->id, $commands); + $this->fail('A failed staged put was accepted.'); + } catch (RuntimeException $error) { + $this->assertSame('put failed', $error->getMessage()); + } + + Event::assertNotDispatched(WorkflowCompleted::class); + $this->assertSame([], $storage->puts); + $this->assertNothingRecorded($run); + $this->assertSame($before, $this->leaseOf($task)); + $this->assertSame(RunStatus::Waiting, $run->fresh()->status); + + $storage->failPuts = false; + $accepted = $bridge->complete($task->id, $commands); + + $this->assertTrue($accepted['completed'], json_encode($accepted, JSON_THROW_ON_ERROR)); + $this->assertCount(1, $storage->puts); + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + } + + /** + * Workflow #930: the walk back from completion is checked against the + * earliest storable timestamp before anything is written. + */ + public function testAReportReachingBeforeTheEarliestStorableTimestampIsRefused(): void + { + Carbon::setTestNow('1970-01-02T00:00:00.000000Z'); + + try { + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $command = $this->localActivity(0); + $command['attempts'][0]['duration_ms'] = 2 * 86_400_000; + + $refused = $this->app->make(WorkflowTaskBridge::class)->complete($task->id, [$command]); + + $this->assertSame('local_activity_timestamp_out_of_range', $refused['reason']); + $this->assertNothingRecorded($run); + } finally { + Carbon::setTestNow(); + } + } + + /** + * Workflow #930: the meter charges exactly the rows a local activity + * writes: each event and the run position it advances, the execution, + * each attempt and the failure; the preflight counts the same rows. + */ + public function testTheMeterChargesTheRowsALocalActivityWrites(): void + { + $run = $this->createWaitingRun(); + $task = $this->createLeasedTask($run); + $command = [ + ...$this->localActivity(2), + 'outcome' => 'failed', + 'message' => 'gave up', + 'retry_policy' => [ + 'max_attempts' => 2, + ], + 'attempts' => [ + [ + 'attempt_number' => 1, + 'outcome' => 'failed', + 'message' => 'first', + 'retry_reason' => 'failure', + 'backoff_seconds' => 0, + 'heartbeats' => [[ + 'elapsed_ms' => 1, + ], [ + 'elapsed_ms' => 2, + ]], + ], + [ + 'attempt_number' => 2, + 'outcome' => 'failed', + 'message' => 'gave up', + 'heartbeats' => [], + ], + ], + ]; + unset($command['result']); + + $result = $this->app->make(WorkflowTaskBridge::class)->complete($task->id, [$command]); + + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $events = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->whereIn('event_type', [ + 'ActivityScheduled', 'ActivityStarted', 'ActivityHeartbeatRecorded', 'ActivityRetryScheduled', 'ActivityFailed', + ])->count(); + $normalized = WorkflowCommandNormalizer::normalize([$command], '1.19')[0]; + $this->assertSame(WorkflowCommandNormalizer::localActivityHistoryEvents($normalized), $events); + $inserted = 1 + ActivityAttempt::query()->where('workflow_run_id', $run->id)->count() + + \Workflow\V2\Models\WorkflowFailure::query()->where('workflow_run_id', $run->id)->count(); + $this->assertSame(WorkflowCommandNormalizer::localActivityRows($normalized), $inserted); + $used = HistoryTransactionMeter::lastUsed(); + // The local activity's rows are part of what the meter charged. + $this->assertGreaterThanOrEqual(2 * $events + $inserted, $used[HistoryTransactionMeter::ROWS]); + $this->assertSame($used[HistoryTransactionMeter::EVENTS] * 2 + $inserted, $used[HistoryTransactionMeter::ROWS]); + } + + /** + * Workflow #930: a run's deadline close is not the worker's work and is + * not metered, so a run whose close exceeds one completion's budget still + * times out instead of refusing every completion. + */ + public function testADeadlineCloseIsNotBlockedByTheCompletionBudget(): void + { + config([ + 'workflows.v2.structural_limits.history_transaction_size' => 1, + ]); + $run = $this->createWaitingRun(); + $run->forceFill([ + 'execution_deadline_at' => now() + ->subMinute(), + ])->save(); + foreach (range(1, 3) as $sequence) { + WorkflowTimer::query()->create([ + 'workflow_run_id' => $run->id, + 'sequence' => $sequence, + 'status' => TimerStatus::Pending->value, + 'delay_seconds' => 60, + 'fire_at' => now() + ->addMinute(), + ]); + } + $task = $this->createLeasedTask($run); + + $result = $this->app->make(WorkflowTaskBridge::class)->complete($task->id, [$this->localActivity(0)]); + + $this->assertSame('run_timed_out', $result['reason'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame(RunStatus::Failed, $run->fresh()->status); + // Three timer cancellations and the timeout: more than the budget of one. + $this->assertGreaterThan(1, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + } + + /** + * @return array + */ + private function localActivity(int $heartbeats): array + { + return [ + 'type' => 'record_local_activity', + 'activity_type' => 'metered-local-activity', + 'result' => Serializer::serialize('done'), + 'outcome' => 'completed', + 'attempts' => [[ + 'attempt_number' => 1, + 'outcome' => 'completed', + 'heartbeats' => array_map( + static fn (int $elapsed): array => [ + 'elapsed_ms' => $elapsed, + ], + $heartbeats === 0 ? [] : range(1, $heartbeats) + ), + ]], + 'execution_mode' => 'local', + ]; + } + + /** + * @return array + */ + private function failedLocalActivity(): array + { + $command = [ + ...$this->localActivity(0), + 'outcome' => 'failed', + 'message' => 'gave up', + ]; + $command['attempts'][0] = [ + 'attempt_number' => 1, + 'outcome' => 'failed', + 'message' => 'gave up', + 'heartbeats' => [], + ]; + unset($command['result']); + + return $command; + } + + /** + * @return list + */ + private function leaseOf(WorkflowTask $task): array + { + $task = $task->fresh(); + + return [$task->status, $task->lease_owner, $task->lease_expires_at?->toJSON(), $task->attempt_count]; + } + + private function assertNothingRecorded(WorkflowRun $run): void + { + $this->assertSame(0, ActivityExecution::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(0, ActivityAttempt::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(0, WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(0, $run->fresh()->last_history_sequence ?? 0); + } + private function createWaitingRun(): WorkflowRun { $instance = WorkflowInstance::query()->create([ diff --git a/tests/Unit/V2/ExternalPayloadStorageTest.php b/tests/Unit/V2/ExternalPayloadStorageTest.php index e102d027..48731e14 100644 --- a/tests/Unit/V2/ExternalPayloadStorageTest.php +++ b/tests/Unit/V2/ExternalPayloadStorageTest.php @@ -10,6 +10,7 @@ use League\Flysystem\Local\LocalFilesystemAdapter; use PHPUnit\Framework\TestCase; use Workflow\V2\Contracts\ExternalPayloadStorageDriver; +use Workflow\V2\Contracts\PredictableExternalPayloadStorageDriver; use Workflow\V2\Exceptions\ExternalPayloadIntegrityException; use Workflow\V2\Support\ExternalPayloadReference; use Workflow\V2\Support\ExternalPayloadStorage; @@ -150,6 +151,105 @@ public function testLocalFilesystemDriverRejectsUrisOutsideRoot(): void $driver->get('file:///etc/passwd'); } + public function testLocalDriverNeverMovesAnIncompleteOrCorruptWriteIntoPlace(): void + { + $root = $this->makeStorageRoot(); + $data = 'encoded-payload'; + $sha256 = hash('sha256', $data); + $writers = [ + 'short write' => static fn (string $path, string $content): int|false => file_put_contents( + $path, + substr($content, 0, 5) + ), + 'short write reported whole' => static function (string $path, string $content): int { + file_put_contents($path, substr($content, 0, 5)); + + return strlen($content); + }, + 'same length, wrong bytes' => static fn (string $path, string $content): int|false => file_put_contents( + $path, + strrev($content) + ), + 'failed write' => static fn (string $path, string $content): int|false => false, + ]; + + foreach ($writers as $case => $writer) { + $driver = new LocalFilesystemExternalPayloadStorage($root, $writer); + try { + $driver->put($data, $sha256, 'avro'); + $this->fail("{$case}: the incomplete write was accepted."); + } catch (ExternalPayloadIntegrityException $error) { + $this->assertStringContainsString('was not written completely', $error->getMessage(), $case); + } + $path = $this->pathFromFileUri($driver->uriFor($sha256, 'avro')); + $this->assertFileDoesNotExist($path, $case); + $this->assertSame([], glob(dirname($path) . '/*.tmp') ?: [], $case); + } + } + + public function testLocalDriverKeepsAMatchingObjectAndReplacesAPartialOrCorruptOne(): void + { + $root = $this->makeStorageRoot(); + $data = 'encoded-payload'; + $sha256 = hash('sha256', $data); + $writes = 0; + $driver = new LocalFilesystemExternalPayloadStorage( + $root, + static function (string $path, string $content) use (&$writes): int|false { + $writes++; + + return file_put_contents($path, $content); + } + ); + + $uri = $driver->put($data, $sha256, 'avro'); + $this->assertSame($driver->uriFor($sha256, 'avro'), $uri); + $this->assertSame(1, $writes); + $path = $this->pathFromFileUri($uri); + + // Idempotent: a matching object is kept as it is. + $this->assertSame($uri, $driver->put($data, $sha256, 'avro')); + $this->assertSame(1, $writes); + + foreach (['encoded-pay', 'encoded-payloae'] as $damaged) { + file_put_contents($path, $damaged); + $this->assertSame($uri, $driver->put($data, $sha256, 'avro')); + $this->assertSame($data, file_get_contents($path)); + } + $this->assertSame(3, $writes); + $this->assertSame([], glob(dirname($path) . '/*.tmp') ?: []); + } + + public function testLocalDriverRefusesContentThatDoesNotMatchItsHash(): void + { + $driver = new LocalFilesystemExternalPayloadStorage($this->makeStorageRoot()); + $sha256 = hash('sha256', 'other-payload'); + + try { + $driver->put('encoded-payload', $sha256, 'avro'); + $this->fail('Content that does not match its hash was stored.'); + } catch (ExternalPayloadIntegrityException) { + } + + $this->assertFileDoesNotExist($this->pathFromFileUri($driver->uriFor($sha256, 'avro'))); + } + + public function testOnlyTheLocalDriverPromisesPredictableAtomicStorage(): void + { + $root = $this->makeStorageRoot(); + + $this->assertInstanceOf( + PredictableExternalPayloadStorageDriver::class, + new LocalFilesystemExternalPayloadStorage($root) + ); + // A filesystem disk does not guarantee an atomic write, so this driver + // keeps the original interface and stores payloads as before. + $this->assertNotInstanceOf( + PredictableExternalPayloadStorageDriver::class, + new FilesystemExternalPayloadStorage($this->makeDisk($root), 's3', 'workflow-payloads') + ); + } + public function testFilesystemDriverStoresFetchesAndDeletesVerifiedBytes(): void { $driver = new FilesystemExternalPayloadStorage( diff --git a/tests/Unit/V2/HistoryTransactionMeterTest.php b/tests/Unit/V2/HistoryTransactionMeterTest.php new file mode 100644 index 00000000..54bb027d --- /dev/null +++ b/tests/Unit/V2/HistoryTransactionMeterTest.php @@ -0,0 +1,390 @@ +chargeEvent('ActivityStarted', [ + 'a' => 'b/é', + ]); + $meter->chargeRows(2); + + return $meter->used(); + }); + + // {"a":"b/é"}: slashes and unicode unescaped, as HistoryBudget stores it. + $this->assertSame([ + HistoryTransactionMeter::EVENTS => 1, + HistoryTransactionMeter::BYTES => strlen('ActivityStarted') + strlen('{"a":"b/é"}'), + HistoryTransactionMeter::ROWS => 3, + HistoryTransactionMeter::STAGED_BYTES => 0, + ], $used); + } + + public function testEachLimitAllowsItsExactValueAndRefusesOneMore(): void + { + foreach ([ + HistoryTransactionMeter::EVENTS => 'history_transaction_size', + HistoryTransactionMeter::ROWS => 'history_transaction_rows', + ] as $dimension => $config) { + config([ + 'workflows.v2.structural_limits.' . $config => 3, + ]); + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($dimension): void { + $dimension === HistoryTransactionMeter::EVENTS + ? array_map(static fn () => $meter->chargeEvent('E', []), range(1, 3)) + : $meter->chargeRows(3); + }); + try { + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($dimension): void { + $dimension === HistoryTransactionMeter::EVENTS + ? array_map(static fn () => $meter->chargeEvent('E', []), range(1, 4)) + : $meter->chargeRows(4); + }); + $this->fail("{$dimension} accepted one over its limit."); + } catch (HistoryTransactionLimitExceededException $refusal) { + $this->assertSame($dimension, $refusal->dimension); + $this->assertSame(4, $refusal->used); + $this->assertSame(3, $refusal->limit); + } + config([ + 'workflows.v2.structural_limits.' . $config => null, + ]); + } + + // A disabled limit (0) charges without refusing. + config([ + 'workflows.v2.structural_limits.history_transaction_bytes' => 0, + ]); + $used = HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter): int { + $meter->chargeEvent('E', [ + 'data' => str_repeat('x', 100_000), + ]); + + return $meter->used()[HistoryTransactionMeter::BYTES]; + }); + $this->assertGreaterThan(100_000, $used); + } + + public function testACaughtRefusalCannotContinueToACommit(): void + { + config([ + 'workflows.v2.structural_limits.history_transaction_rows' => 1, + ]); + $this->expectException(HistoryTransactionLimitExceededException::class); + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter): string { + try { + $meter->chargeRows(2); + } catch (HistoryTransactionLimitExceededException) { + // Swallowed by mistake: the meter still refuses before returning. + } + + return 'committed'; + }); + } + + public function testAggregatesAreOverflowChecked(): void + { + $this->assertSame(PHP_INT_MAX, HistoryTransactionMeter::checkedAdd(PHP_INT_MAX - 1, 1)); + $this->expectException(HistoryTransactionLimitExceededException::class); + HistoryTransactionMeter::checkedAdd(PHP_INT_MAX, 1); + } + + public function testMultiplicationIsOverflowChecked(): void + { + $this->assertSame(6, HistoryTransactionMeter::checkedMultiply(2, 3)); + $this->expectException(HistoryTransactionLimitExceededException::class); + HistoryTransactionMeter::checkedMultiply(PHP_INT_MAX, 2); + } + + public function testAnUnmeasurableEventFailsClosed(): void + { + try { + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter): void { + $meter->chargeEvent('E', [ + 'bad' => "\xB1\x31", + ]); + }); + $this->fail('An event json_encode cannot measure was accepted.'); + } catch (HistoryTransactionLimitExceededException $refusal) { + $this->assertSame(HistoryTransactionLimitExceededException::MEASURE_UNAVAILABLE, $refusal->dimension); + } + } + + public function testStagedPayloadsAreStoredOnlyWhenTheWholeAttemptPasses(): void + { + $driver = new RecordingPredictableDriver(); + + // Refused after staging: nothing is stored. + config([ + 'workflows.v2.structural_limits.history_transaction_rows' => 1, + ]); + try { + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($driver): void { + $meter->stage($driver, 'payload-one', 'avro'); + $meter->chargeRows(2); + }); + $this->fail('A refused attempt returned.'); + } catch (HistoryTransactionLimitExceededException) { + } + $this->assertSame([], $driver->puts); + + // Accepted: each distinct object once, under its predicted URI, while + // every occurrence counts toward the staged bytes. + config([ + 'workflows.v2.structural_limits.history_transaction_rows' => null, + ]); + $used = HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($driver): int { + $first = $meter->stage($driver, 'payload-one', 'avro'); + $meter->stage($driver, 'payload-one', 'avro'); + $meter->stage($driver, 'payload-two', 'avro'); + TestCaseAssert::same($driver->uriFor(hash('sha256', 'payload-one'), 'avro'), $first->uri); + + return $meter->used()[HistoryTransactionMeter::STAGED_BYTES]; + }); + $this->assertSame(3 * strlen('payload-one'), $used); + $this->assertSame( + [hash('sha256', 'payload-one'), hash('sha256', 'payload-two')], + array_column($driver->puts, 'sha256') + ); + + // A fresh attempt starts with an empty stage. + $driver->puts = []; + HistoryTransactionMeter::run( + static fn (HistoryTransactionMeter $meter): int => $meter->used()[HistoryTransactionMeter::STAGED_BYTES] + ); + $this->assertSame([], $driver->puts); + } + + public function testRawStagedBytesAreLimitedPerPayloadAndPerTransaction(): void + { + $driver = new RecordingPredictableDriver(); + config([ + 'workflows.v2.structural_limits.payload_size_bytes' => 10, + 'workflows.v2.structural_limits.external_payload_transaction_bytes' => 20, + ]); + + foreach ([ + [[str_repeat('a', 11)], HistoryTransactionMeter::PAYLOAD_SIZE], + [[str_repeat('a', 10), str_repeat('b', 10), 'c'], HistoryTransactionMeter::STAGED_BYTES], + ] as [$payloads, $dimension]) { + try { + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ( + $driver, + $payloads + ): void { + foreach ($payloads as $payload) { + $meter->stage($driver, $payload, 'avro'); + } + }); + $this->fail("{$dimension} accepted one byte over."); + } catch (HistoryTransactionLimitExceededException $refusal) { + $this->assertSame($dimension, $refusal->dimension); + } + } + $this->assertSame([], $driver->puts); + + // Exactly at both limits. + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($driver): void { + $meter->stage($driver, str_repeat('a', 10), 'avro'); + $meter->stage($driver, str_repeat('b', 10), 'avro'); + }); + $this->assertCount(2, $driver->puts); + } + + public function testAMispredictedUriOrAFailedPutRollsTheAttemptBack(): void + { + $lying = new RecordingPredictableDriver(lie: true); + try { + HistoryTransactionMeter::run( + static fn (HistoryTransactionMeter $meter) => $meter->stage($lying, 'x', 'avro') + ); + $this->fail('A mispredicted URI was accepted.'); + } catch (HistoryTransactionLimitExceededException $refusal) { + $this->assertSame(HistoryTransactionLimitExceededException::STAGED_URI_MISMATCH, $refusal->dimension); + } + + $failing = new RecordingPredictableDriver(failAfter: 1); + try { + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($failing): void { + $meter->stage($failing, 'first', 'avro'); + $meter->stage($failing, 'second', 'avro'); + }); + $this->fail('A failed put was accepted.'); + } catch (RuntimeException $error) { + $this->assertSame('put failed', $error->getMessage()); + } + // The first object stays (content-addressed objects may be shared); + // the attempt itself failed and rolls back. + $this->assertCount(1, $failing->puts); + } + + public function testOneUriForDifferentPayloadsFailsClosedAndStoresNothing(): void + { + foreach ([ + [['first', 'avro'], ['second', 'avro']], + [['same', 'avro'], ['same', 'json']], + ] as [$first, $second]) { + $colliding = new RecordingPredictableDriver(collide: true); + try { + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ( + $colliding, + $first, + $second + ): void { + $meter->stage($colliding, ...$first); + $meter->stage($colliding, ...$second); + }); + $this->fail('One predicted URI was accepted for two different payloads.'); + } catch (HistoryTransactionLimitExceededException $refusal) { + $this->assertSame(HistoryTransactionLimitExceededException::STAGED_URI_COLLISION, $refusal->dimension); + } + $this->assertSame([], $colliding->puts); + } + + // The same payload staged twice through one driver is one object. + $colliding = new RecordingPredictableDriver(collide: true); + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($colliding): void { + $meter->stage($colliding, 'same', 'avro'); + $meter->stage($colliding, 'same', 'avro'); + }); + $this->assertCount(1, $colliding->puts); + } + + public function testTwoDriversPredictingOneUriEachStoreTheirObject(): void + { + $left = new RecordingPredictableDriver(collide: true); + $right = new RecordingPredictableDriver(collide: true); + + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($left, $right): void { + $meter->stage($left, 'left', 'avro'); + $meter->stage($right, 'right', 'avro'); + }); + + $this->assertSame([hash('sha256', 'left')], array_column($left->puts, 'sha256')); + $this->assertSame([hash('sha256', 'right')], array_column($right->puts, 'sha256')); + } + + public function testUnstagedPayloadsAreChargedAgainstTheSameLimits(): void + { + config([ + 'workflows.v2.structural_limits.payload_size_bytes' => 10, + 'workflows.v2.structural_limits.external_payload_transaction_bytes' => 20, + ]); + + $used = HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter): int { + $meter->chargeUnstaged(10); + $meter->chargeUnstaged(10); + + return $meter->used()[HistoryTransactionMeter::STAGED_BYTES]; + }); + $this->assertSame(20, $used); + + foreach ([[11], [10, 10, 1]] as $sizes) { + try { + HistoryTransactionMeter::run(static function (HistoryTransactionMeter $meter) use ($sizes): void { + foreach ($sizes as $size) { + $meter->chargeUnstaged($size); + } + }); + $this->fail('An unstaged payload past a limit was accepted.'); + } catch (HistoryTransactionLimitExceededException $refusal) { + $this->assertSame( + $sizes === [11] ? HistoryTransactionMeter::PAYLOAD_SIZE : HistoryTransactionMeter::STAGED_BYTES, + $refusal->dimension + ); + } + } + } + + public function testMetersDoNotNest(): void + { + $this->expectException(LogicException::class); + HistoryTransactionMeter::run(static fn () => HistoryTransactionMeter::run(static fn () => null)); + } + + public function testAFailedRunLeavesNoCurrentMeter(): void + { + try { + HistoryTransactionMeter::run(static function (): void { + throw new RuntimeException('body failed'); + }); + } catch (RuntimeException) { + } + + $this->assertNull(HistoryTransactionMeter::current()); + } +} + +/** + * Assertions callable from static closures. + */ +final class TestCaseAssert +{ + public static function same(mixed $expected, mixed $actual): void + { + \PHPUnit\Framework\Assert::assertSame($expected, $actual); + } +} + +/** + * A content-addressed driver that records every put. A colliding driver + * predicts one URI for every payload. + */ +final class RecordingPredictableDriver implements PredictableExternalPayloadStorageDriver +{ + /** + * @var list + */ + public array $puts = []; + + public function __construct( + private readonly bool $lie = false, + private readonly ?int $failAfter = null, + private readonly bool $collide = false + ) { + } + + public function uriFor(string $sha256, string $codec): string + { + return $this->collide ? 'memory://one-uri' : "memory://{$codec}/{$sha256}"; + } + + public function put(string $data, string $sha256, string $codec): string + { + if ($this->failAfter !== null && count($this->puts) >= $this->failAfter) { + throw new RuntimeException('put failed'); + } + $this->puts[] = [ + 'sha256' => $sha256, + 'codec' => $codec, + ]; + + return $this->lie ? "memory://elsewhere/{$sha256}" : $this->uriFor($sha256, $codec); + } + + public function get(string $uri): string + { + throw new RuntimeException('not used'); + } + + public function delete(string $uri): void + { + throw new RuntimeException('not used'); + } +} diff --git a/tests/Unit/V2/LifecycleEventDispatcherCollectTest.php b/tests/Unit/V2/LifecycleEventDispatcherCollectTest.php new file mode 100644 index 00000000..015507ea --- /dev/null +++ b/tests/Unit/V2/LifecycleEventDispatcherCollectTest.php @@ -0,0 +1,223 @@ +createRun(); + $attempt = 0; + + [$result, $events] = DB::transaction(static function () use (&$attempt, $run): array { + return LifecycleEventDispatcher::collect(static function () use (&$attempt, $run): string { + LifecycleEventDispatcher::workflowCompleted($run); + if (++$attempt === 1) { + // Laravel retries a transaction attempt that fails with a deadlock. + throw new RuntimeException('Deadlock found when trying to get lock; try restarting transaction'); + } + + return 'committed'; + }); + }, 2); + + $this->assertSame(2, $attempt); + $this->assertSame('committed', $result); + Event::assertNotDispatched(WorkflowCompleted::class); + + LifecycleEventDispatcher::release($events); + + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + } + + public function testAFailedAttemptDispatchesNothing(): void + { + Event::fake([WorkflowCompleted::class]); + $run = $this->createRun(); + + $body = static function () use ($run): void { + LifecycleEventDispatcher::workflowCompleted($run); + + throw new RuntimeException('rolled back'); + }; + + try { + DB::transaction(static fn (): array => LifecycleEventDispatcher::collect($body)); + $this->fail('The failed attempt returned.'); + } catch (RuntimeException $error) { + $this->assertSame('rolled back', $error->getMessage()); + } + + // A later dispatch outside any collecting attempt is immediate. + Event::assertNotDispatched(WorkflowCompleted::class); + LifecycleEventDispatcher::workflowCompleted($run); + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + } + + public function testEventsReleasedInsideAnOpenTransactionWaitForItsCommit(): void + { + Event::fake([WorkflowCompleted::class]); + $run = $this->createRun(); + + DB::transaction(static function () use ($run): void { + [, $events] = LifecycleEventDispatcher::collect( + static fn () => LifecycleEventDispatcher::workflowCompleted($run) + ); + LifecycleEventDispatcher::release($events); + Event::assertNotDispatched(WorkflowCompleted::class); + }); + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + + try { + DB::transaction(static function () use ($run): void { + [, $events] = LifecycleEventDispatcher::collect( + static fn () => LifecycleEventDispatcher::workflowCompleted($run) + ); + LifecycleEventDispatcher::release($events); + + throw new RuntimeException('outer rolled back'); + }); + } catch (RuntimeException) { + } + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + } + + public function testANestedAttemptThatRollsBackPublishesNothingWhenTheOuterCommits(): void + { + Event::fake([WorkflowCompleted::class, WorkflowFailed::class]); + $run = $this->createRun(); + + [$result, $events] = DB::transaction(static fn (): array => LifecycleEventDispatcher::collect( + static function () use ($run): string { + LifecycleEventDispatcher::workflowCompleted($run); + try { + DB::transaction(static fn (): array => LifecycleEventDispatcher::collect(static function () use ( + $run + ): void { + LifecycleEventDispatcher::workflowFailed($run, RuntimeException::class, 'nested'); + + throw new RuntimeException('nested attempt rolled back'); + })); + } catch (RuntimeException) { + // The outer attempt carries on and commits. + } + + return 'outer committed'; + } + )); + LifecycleEventDispatcher::release($events); + + $this->assertSame('outer committed', $result); + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + Event::assertNotDispatched(WorkflowFailed::class); + } + + public function testARetriedNestedTransactionPublishesOnlyItsCommittedAttempt(): void + { + Event::fake([WorkflowCompleted::class]); + $run = $this->createRun(); + + [, $events] = DB::transaction(static fn (): array => LifecycleEventDispatcher::collect( + static function () use ($run): void { + for ($attempt = 1; ; $attempt++) { + try { + [, $nested] = DB::transaction( + static fn (): array => LifecycleEventDispatcher::collect(static function () use ( + $run, + $attempt + ): void { + LifecycleEventDispatcher::workflowCompleted($run); + if ($attempt === 1) { + throw new RuntimeException('nested attempt failed'); + } + }) + ); + LifecycleEventDispatcher::release($nested); + + return; + } catch (RuntimeException) { + // Retry the nested transaction. + } + } + } + )); + + // The committed nested attempt's event joined the outer attempt. + Event::assertNotDispatched(WorkflowCompleted::class); + $this->assertCount(1, $events); + LifecycleEventDispatcher::release($events); + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + } + + public function testAnEnclosingAttemptThatFailsDiscardsItsNestedEvents(): void + { + Event::fake([WorkflowCompleted::class]); + $run = $this->createRun(); + + try { + LifecycleEventDispatcher::collect(static function () use ($run): void { + [, $nested] = LifecycleEventDispatcher::collect( + static fn () => LifecycleEventDispatcher::workflowCompleted($run) + ); + LifecycleEventDispatcher::release($nested); + + throw new RuntimeException('outer attempt failed'); + }); + } catch (RuntimeException) { + } + + Event::assertNotDispatched(WorkflowCompleted::class); + // Nothing is left collecting: a later dispatch is immediate. + LifecycleEventDispatcher::workflowCompleted($run); + Event::assertDispatchedTimes(WorkflowCompleted::class, 1); + } + + private function createRun(): WorkflowRun + { + $instance = WorkflowInstance::query()->create([ + 'workflow_class' => TestGreetingWorkflow::class, + 'workflow_type' => 'test-greeting-workflow', + 'run_count' => 1, + 'started_at' => now() + ->subMinute(), + ]); + $run = WorkflowRun::query()->create([ + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'workflow_class' => TestGreetingWorkflow::class, + 'workflow_type' => 'test-greeting-workflow', + 'status' => RunStatus::Completed->value, + 'arguments' => Serializer::serialize(['Taylor']), + 'connection' => 'redis', + 'queue' => 'default', + 'compatibility' => 'build-a', + 'started_at' => now() + ->subMinute(), + ]); + $instance->forceFill([ + 'current_run_id' => $run->id, + ])->save(); + + return $run; + } +} diff --git a/tests/Unit/V2/LocalActivityCommandNormalizationTest.php b/tests/Unit/V2/LocalActivityCommandNormalizationTest.php index 6334f52f..e3c9ef31 100644 --- a/tests/Unit/V2/LocalActivityCommandNormalizationTest.php +++ b/tests/Unit/V2/LocalActivityCommandNormalizationTest.php @@ -338,6 +338,143 @@ public function testExplicitNullOptionalAttemptStringsAreOmittedWithoutChangingT $this->assertSame($original, $command); } + /** + * Workflow #930: the timing fields recording walks back from completion + * are bounded, so no report can drive timestamp arithmetic out of range. + */ + #[DataProvider('timingFields')] + public function testTimingFieldsAreBoundedForTimestampArithmetic(string $field, int $maximum, string $message): void + { + $first = $this->retriedCommand($field, $maximum)['attempts'][0]; + $this->assertSame($maximum, $field === 'elapsed_ms' ? $first['heartbeats'][0]['elapsed_ms'] : $first[$field]); + foreach ([$maximum + 1, PHP_INT_MAX] as $value) { + try { + $this->retriedCommand($field, $value); + } catch (ValidationException $error) { + $path = $field === 'elapsed_ms' ? 'commands.0.attempts.0.heartbeats.0.elapsed_ms' : 'commands.0.attempts.0.' . $field; + $this->assertSame([ + $path => [$message], + ], $error->errors()); + + continue; + } + + $this->fail(sprintf('%s %d was accepted.', $field, $value)); + } + } + + /** + * @return iterable + */ + public static function timingFields(): iterable + { + yield 'attempt duration' => ['duration_ms', WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_DURATION_MS, + 'Local activity attempt duration_ms must not exceed 604800000.']; + yield 'heartbeat offset' => ['elapsed_ms', WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_DURATION_MS, + 'Local activity heartbeat elapsed_ms must not exceed 604800000.']; + yield 'retry backoff' => ['backoff_seconds', WorkflowCommandNormalizer::MAX_LOCAL_ACTIVITY_BACKOFF_SECONDS, + 'Local activity attempt backoff_seconds must not exceed 86400.']; + } + + public function testACompletionRecordsAtMostOneHistoryTransactionOfEvents(): void + { + $this->assertSame(5000, \Workflow\V2\Support\StructuralLimits::historyTransactionSizeLimit()); + $normalize = static fn (array $commands): array => WorkflowCommandNormalizer::normalize($commands, '1.19'); + // Four commands of one attempt and 1,000 heartbeats record 1,003 + // events each; a fifth of one attempt records 3 plus its heartbeats. + $commands = array_fill(0, 4, $this->heartbeatCommand(1000)); + $this->assertSame(1003, WorkflowCommandNormalizer::localActivityHistoryEvents($this->heartbeatCommand(1000))); + $this->assertSame([], WorkflowCommandNormalizer::completionBudgetErrors( + $normalize([...$commands, $this->heartbeatCommand(985)]), + )); + $this->assertSame([ + 'This completion would record more than 5000 history events; one workflow task completion may record at most 5000.', + ], WorkflowCommandNormalizer::completionBudgetErrors($normalize([...$commands, $this->heartbeatCommand(986)]))); + // Every other command counts as one event, a lower bound. + $this->assertSame(7, WorkflowCommandNormalizer::localActivityHistoryEvents([ + 'attempts' => [[ + 'heartbeats' => [[], []], + ], [ + 'heartbeats' => [], + ]], + ])); + config([ + 'workflows.v2.structural_limits.history_transaction_size' => 4, + ]); + $this->assertSame([], WorkflowCommandNormalizer::completionBudgetErrors( + $normalize([[ + 'type' => 'complete_workflow', + ], $this->heartbeatCommand(0)]), + )); + $this->assertCount(1, WorkflowCommandNormalizer::completionBudgetErrors( + $normalize([[ + 'type' => 'complete_workflow', + ], $this->heartbeatCommand(1)]), + )); + // A disabled limit (0) refuses nothing. + config([ + 'workflows.v2.structural_limits.history_transaction_size' => 0, + ]); + $this->assertSame( + [], + WorkflowCommandNormalizer::completionBudgetErrors($normalize([...$commands, $commands[0]])) + ); + } + + /** + * A failed first attempt retried into a completed second, with the timing + * field under test set on the first. + * + * @return array + */ + private function retriedCommand(string $field, int $value): array + { + $first = [ + 'attempt_number' => 1, + 'outcome' => 'failed', + 'message' => 'declined', + 'retry_reason' => 'failure', + 'backoff_seconds' => 0, + 'heartbeats' => [], + ]; + if ($field === 'elapsed_ms') { + $first['heartbeats'] = [[ + 'elapsed_ms' => $value, + ]]; + } else { + $first[$field] = $value; + } + + return WorkflowCommandNormalizer::normalize([[ + ...$this->completedCommand(), + 'retry_policy' => [ + 'max_attempts' => 2, + ], + 'attempts' => [ + $first, [ + 'attempt_number' => 2, + 'outcome' => 'completed', + 'heartbeats' => [], + ]], + ]], '1.19')[0]; + } + + /** + * @return array + */ + private function heartbeatCommand(int $heartbeats): array + { + $command = $this->completedCommand(); + $command['attempts'][0]['heartbeats'] = array_map( + static fn (int $elapsed): array => [ + 'elapsed_ms' => $elapsed, + ], + $heartbeats === 0 ? [] : range(0, $heartbeats - 1), + ); + + return $command; + } + /** * @return array */ diff --git a/tests/Unit/V2/StructuralLimitsTest.php b/tests/Unit/V2/StructuralLimitsTest.php index 4bf7371c..a125b824 100644 --- a/tests/Unit/V2/StructuralLimitsTest.php +++ b/tests/Unit/V2/StructuralLimitsTest.php @@ -9,6 +9,9 @@ use Tests\TestCase; use Workflow\V2\Enums\StructuralLimitKind; use Workflow\V2\Exceptions\StructuralLimitExceededException; +use Workflow\V2\Support\BackendCapabilities; +use Workflow\V2\Support\HealthCheck; +use Workflow\V2\Support\OperatorMetrics; use Workflow\V2\Support\StructuralLimits; final class StructuralLimitsTest extends TestCase @@ -644,9 +647,40 @@ public function testSnapshotReturnsAllLimits(): void $this->assertArrayHasKey('memo_size_bytes', $snapshot); $this->assertArrayHasKey('search_attribute_size_bytes', $snapshot); $this->assertArrayHasKey('history_transaction_size', $snapshot); + $this->assertArrayHasKey('history_transaction_bytes', $snapshot); + $this->assertArrayHasKey('history_transaction_rows', $snapshot); + $this->assertArrayHasKey('external_payload_transaction_bytes', $snapshot); $this->assertArrayHasKey('warning_threshold_percent', $snapshot); } + public function testCompletionBudgetLimitsArePublishedOnEveryOperatorSurface(): void + { + $configured = [ + 'history_transaction_bytes' => 123_456, + 'history_transaction_rows' => 7_890, + 'external_payload_transaction_bytes' => 654_321, + ]; + foreach ($configured as $key => $value) { + config([ + 'workflows.v2.structural_limits.' . $key => $value, + ]); + } + + $capabilities = BackendCapabilities::snapshot()['structural_limits']; + $surfaces = [ + 'StructuralLimits::snapshot' => StructuralLimits::snapshot(), + 'HealthCheck::snapshot' => HealthCheck::snapshot()['structural_limits'], + 'OperatorMetrics::snapshot' => OperatorMetrics::snapshot()['structural_limits'], + 'BackendCapabilities configured' => $capabilities['configured'], + 'BackendCapabilities effective' => $capabilities['effective'], + ]; + foreach ($surfaces as $surface => $limits) { + foreach ($configured as $key => $value) { + $this->assertSame($value, $limits[$key] ?? null, "{$surface} must publish {$key}."); + } + } + } + // --------------------------------------------------------------- // StructuralLimitKind enum // --------------------------------------------------------------- @@ -664,6 +698,9 @@ public function testEnumContainsAllLimitKinds(): void 'pending_signal_count', 'pending_update_count', 'command_batch_size', + 'history_transaction_bytes', + 'history_transaction_rows', + 'external_payload_transaction_bytes', ]; $actual = array_map(