Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,12 @@

## Unreleased

- Add an opt-in first-open wall deadline for condition-wait commands. With
`deadline_policy: "occurrence"`, physical false-wake reopens retain the first
timer's exact deadline and original timeout budget. Conflicting, timed-out
or unprovable chains refuse with whole-completion rollback. Commands without
the policy retain their existing per-open behavior.

## 2.21.1 - 2026-11-07

- Exclude the test tree from Composer distribution archives (#1064), avoiding
Expand Down
28 changes: 28 additions & 0 deletions docs/api-stability.md
Original file line number Diff line number Diff line change
Expand Up @@ -988,3 +988,31 @@ on a class in the list, or narrows a return type must either be shipped
in a major version, or promote the class to a `Contracts\*` interface in
the same change. Reviewers should treat unmotivated removals from this
list as a breaking change.

### Occurrence deadlines (worker protocol 1.22)

A runtime advertising `condition_wait_occurrence_deadline` accepts the optional
`deadline_policy: "occurrence"` and
`condition_wait_timeout_budget_seconds` fields on `open_condition_wait`.
The original budget is distinct from the physical open's remaining
`timeout_seconds`; every open of one run/occurrence/key/fingerprint chain keeps
the same policy and original budget. The first open's remaining timeout equals
its full budget. The runtime owns the anchor A, records
`occurrence_deadline_at`, and uses the first timer's exact stored `fire_at` for
every reopen, including an already-due A. It does not fire inline or stamp a
fictitious earlier dispatch time.

Protocol 1.22 describes this additive command/history metadata. Existing request
versions retain their support; this extension adds no new request-version floor
above occurrence identity's 1.17 floor. Writers must observe the runtime
capability before opting in. A recorded legacy chain stays legacy, and a
recorded anchored chain keeps its policy even if the current advertisement
changes. This does not opt grouped waits into the feature. The embedded runner
continues its existing waiting behavior; the opt-in changes command application
for callers that explicitly request this runtime capability.

The entire completion rolls back on `condition_wait_occurrence_mismatch`,
`condition_wait_occurrence_timed_out` or `condition_wait_anchor_unavailable`,
including earlier sibling commands and signal application. Unknown or malformed
policy metadata refuses validation. A missing anchor never starts a fresh
budget. Commands without the policy keep their existing per-open semantics.
12 changes: 12 additions & 0 deletions src/V2/Exceptions/ConditionOccurrenceDeadlineConflict.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
<?php

declare(strict_types=1);

namespace Workflow\V2\Exceptions;

use LogicException;

/** @internal Rolls back the entire completion before reporting a chain refusal. */
final class ConditionOccurrenceDeadlineConflict extends LogicException
{
}
131 changes: 131 additions & 0 deletions src/V2/Support/ConditionOccurrenceDeadline.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
<?php

declare(strict_types=1);

namespace Workflow\V2\Support;

use Carbon\CarbonInterface;
use Workflow\V2\Exceptions\ConditionOccurrenceDeadlineConflict;
use Workflow\V2\Models\WorkflowRun;
use Workflow\V2\Models\WorkflowTimer;

/** @internal The opt-in first-open wall deadline of one authored condition call. */
final class ConditionOccurrenceDeadline
{
/**
* An empty array means legacy; null means invalid. The original budget is
* separate from the physical open's remaining timeout.
*
* @param array<string, mixed> $command
* @return array{deadline_policy: string, condition_wait_timeout_budget_seconds: int}|array{}|null
*/
public static function fields(array $command): ?array
{
$policy = $command['deadline_policy'] ?? null;
$budget = $command['condition_wait_timeout_budget_seconds'] ?? null;
if ($policy === null && $budget === null) {
return [];
}

$timeout = $command['timeout_seconds'] ?? null;
$occurrence = $command['condition_wait_occurrence_id'] ?? null;
if ($policy !== 'occurrence' || ! is_int($budget) || $budget < 0
|| ! is_int($timeout) || $timeout < 0 || $timeout > $budget
|| ! is_string($occurrence) || trim($occurrence) === '') {
return null;
}

return [
'deadline_policy' => $policy,
'condition_wait_timeout_budget_seconds' => $budget,
];
}

/**
* Called under the completion's run lock. Read current history, including
* earlier commands in this batch, rather than a cached relation. A refusal
* is thrown so signal application and every sibling command roll back too.
*
* @param array<string, mixed> $command
* @return array{deadline_at: CarbonInterface|null, keep_open: bool}
*/
public static function resolve(WorkflowRun $run, array $command): array

Check failure on line 52 in src/V2/Support/ConditionOccurrenceDeadline.php

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 24 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=durable-workflow_workflow&issues=AaG5lIMVf0qErduKfGMY&open=AaG5lIMVf0qErduKfGMY&pullRequest=1067

Check warning on line 52 in src/V2/Support/ConditionOccurrenceDeadline.php

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

This method has 4 returns, which is more than the 3 allowed.

See more on https://sonarcloud.io/project/issues?id=durable-workflow_workflow&issues=AaG5lIMVf0qErduKfGMX&open=AaG5lIMVf0qErduKfGMX&pullRequest=1067
{
$fields = self::fields($command);
if ($fields === null) {
throw new ConditionOccurrenceDeadlineConflict('invalid_condition_deadline_policy');
}
$occurrence = $command['condition_wait_occurrence_id'] ?? null;
$budget = $fields['condition_wait_timeout_budget_seconds'] ?? null;
if (! is_string($occurrence) || $occurrence === '') {
return [
'deadline_at' => null,
'keep_open' => false,
];
}

$current = clone $run;
$current->unsetRelation('historyEvents');
$chain = array_values(array_filter(
ConditionWaits::forRun($current),
static fn (array $wait): bool => $wait['condition_wait_occurrence_id'] === $occurrence
));
if ($budget === null) {
foreach ($chain as $wait) {
if ($wait['deadline_policy'] !== null || $wait['condition_wait_timeout_budget_seconds'] !== null) {
throw new ConditionOccurrenceDeadlineConflict('condition_wait_occurrence_mismatch');
}
}

return [
'deadline_at' => null,
'keep_open' => false,
];
}

if ($chain === []) {
if ($command['timeout_seconds'] !== $budget) {
throw new ConditionOccurrenceDeadlineConflict('condition_wait_occurrence_mismatch');
}

return [
'deadline_at' => now()
->addSeconds($budget),
'keep_open' => false,
];
}

$first = $chain[0];
if ($first['deadline_policy'] !== 'occurrence' || $first['condition_wait_timeout_budget_seconds'] === null) {
throw new ConditionOccurrenceDeadlineConflict('condition_wait_anchor_unavailable');
}
$keepOpen = false;
foreach ($chain as $wait) {
if ($wait['deadline_policy'] !== 'occurrence'
|| $wait['condition_wait_timeout_budget_seconds'] !== $budget
|| $wait['condition_key'] !== ($command['condition_key'] ?? null)
|| $wait['condition_definition_fingerprint'] !== ($command['condition_definition_fingerprint'] ?? null)) {
throw new ConditionOccurrenceDeadlineConflict('condition_wait_occurrence_mismatch');
}
if ($wait['timeout_fired_at'] !== null) {
throw new ConditionOccurrenceDeadlineConflict('condition_wait_occurrence_timed_out');
}
$keepOpen = $keepOpen || $wait['status'] === 'open';
}

/** @var WorkflowTimer|null $timer */
$timer = ConfiguredV2Models::query('timer_model', WorkflowTimer::class)
->where('workflow_run_id', $run->id)
->whereKey($first['timer_id'])
->first();
if (! $timer instanceof WorkflowTimer || (int) $timer->sequence !== $first['sequence']
|| ! $timer->fire_at instanceof CarbonInterface) {
throw new ConditionOccurrenceDeadlineConflict('condition_wait_anchor_unavailable');
}

return [
'deadline_at' => $timer->fire_at,
'keep_open' => $keepOpen,
];
}
}
11 changes: 11 additions & 0 deletions src/V2/Support/ConditionWaits.php
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@ final class ConditionWaits
* status: string,
* source_status: string,
* timeout_seconds: int|null,
* deadline_policy: string|null,
* condition_wait_timeout_budget_seconds: int|null,
* timer_id: string|null,
* opened_at: \Carbon\CarbonInterface|null,
* deadline_at: \Carbon\CarbonInterface|null,
Expand Down Expand Up @@ -367,6 +369,8 @@ private static function closeCancelledSelectionWaits(
* status: string,
* source_status: string,
* timeout_seconds: int|null,
* deadline_policy: string|null,
* condition_wait_timeout_budget_seconds: int|null,
* timer_id: string|null,
* opened_at: \Carbon\CarbonInterface|null,
* deadline_at: \Carbon\CarbonInterface|null,
Expand Down Expand Up @@ -400,6 +404,8 @@ private static function openWait(string $waitId, WorkflowHistoryEvent $event): a
* status: string,
* source_status: string,
* timeout_seconds: int|null,
* deadline_policy: string|null,
* condition_wait_timeout_budget_seconds: int|null,
* timer_id: string|null,
* opened_at: \Carbon\CarbonInterface|null,
* deadline_at: \Carbon\CarbonInterface|null,
Expand All @@ -426,6 +432,11 @@ private static function wait(string $waitId, WorkflowHistoryEvent $event): array
'sequence' => self::intValue($event->payload['sequence'] ?? null),
'status' => 'open',
'source_status' => 'waiting',
'deadline_policy' => self::stringValue($event->payload['deadline_policy'] ?? null),
'condition_wait_timeout_budget_seconds' => is_int(
$event->payload['condition_wait_timeout_budget_seconds'] ?? null
)
? $event->payload['condition_wait_timeout_budget_seconds'] : null,
'timeout_seconds' => self::intValue($event->payload['timeout_seconds'] ?? null)
?? self::intValue($event->payload['delay_seconds'] ?? null),
'timer_id' => $timerId,
Expand Down
27 changes: 20 additions & 7 deletions src/V2/Support/DefaultWorkflowTaskBridge.php
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
use Workflow\V2\Enums\TimerStatus;
use Workflow\V2\Enums\UpdateStatus;
use Workflow\V2\Exceptions\ActivityTimeoutException;
use Workflow\V2\Exceptions\ConditionOccurrenceDeadlineConflict;
use Workflow\V2\Exceptions\HistoryEventShapeMismatchException;
use Workflow\V2\Exceptions\HistoryTransactionLimitExceededException;
use Workflow\V2\Exceptions\ServiceOperationAuthorizationException;
Expand Down Expand Up @@ -1509,7 +1510,7 @@ public function completeForProtocol(string $taskId, array $commands, string $pro
'reason' => null,
];
});
} catch (ServiceOperationAuthorizationException $refusal) {
} catch (ServiceOperationAuthorizationException|ConditionOccurrenceDeadlineConflict $refusal) {
return [
'completed' => false,
'task_id' => $taskId,
Expand Down Expand Up @@ -1897,7 +1898,7 @@ private function checkpointCommands(
'reason' => null,
];
}, 5);
} catch (ServiceOperationAuthorizationException $refusal) {
} catch (ServiceOperationAuthorizationException|ConditionOccurrenceDeadlineConflict $refusal) {
return $refused($refusal->getMessage());
} catch (HistoryTransactionLimitExceededException $refusal) {
return $refused(self::meterRefusalReason($refusal));
Expand Down Expand Up @@ -4147,6 +4148,14 @@ private function applyOpenConditionWait(
int $sequence,
array &$createdTaskIds,
): int {
$deadline = ConditionOccurrenceDeadline::resolve($run, $command);
if ($deadline['keep_open']) {
return $sequence;
}
$deadlineMetadata = [
...(ConditionOccurrenceDeadline::fields($command) ?? []),
'occurrence_deadline_at' => $deadline['deadline_at']?->toJSON(),
];
$conditionKey = self::nonEmptyString($command['condition_key'] ?? null);
$conditionDefinitionFingerprint = self::nonEmptyString(
$command['condition_definition_fingerprint'] ?? null,
Expand All @@ -4167,10 +4176,11 @@ private function applyOpenConditionWait(
'condition_definition_fingerprint' => $conditionDefinitionFingerprint,
'sequence' => $sequence,
'timeout_seconds' => $timeoutSeconds,
...$deadlineMetadata,
...$parallelMetadata,
], static fn (mixed $value): bool => $value !== null), $task);

if ($timeoutSeconds === 0) {
if ($timeoutSeconds === 0 && $deadline['deadline_at'] === null) {
$firedEvent = $this->fireImmediateConditionTimeout(
$run,
$task,
Expand Down Expand Up @@ -4198,9 +4208,8 @@ private function applyOpenConditionWait(
$createdTaskIds[] = $resumeTask->id;
}

if ($timeoutSeconds !== null && $timeoutSeconds > 0) {
$fireAt = now()
->addSeconds($timeoutSeconds);
if ($timeoutSeconds !== null && ($timeoutSeconds > 0 || $deadline['deadline_at'] !== null)) {
$fireAt = $deadline['deadline_at'] ?? now()->addSeconds($timeoutSeconds);

/** @var WorkflowTimer $timer */
$timer = WorkflowTimer::query()->create([
Expand All @@ -4217,6 +4226,8 @@ private function applyOpenConditionWait(
'delay_seconds' => $timer->delay_seconds,
'fire_at' => $timer->fire_at?->toJSON(),
'timer_kind' => 'condition_timeout',
'deadline_policy' => $deadlineMetadata['deadline_policy'] ?? null,
'occurrence_deadline_at' => $deadlineMetadata['occurrence_deadline_at'],
'condition_wait_id' => $waitId,
'condition_wait_occurrence_id' => $conditionWaitOccurrenceId,
'condition_key' => $conditionKey,
Expand Down Expand Up @@ -6198,7 +6209,8 @@ private static function normalizeCancelSelectionOperationCommand(array $command)
*/
private static function normalizeOpenConditionWaitCommand(array $command): ?array
{
if (self::hasInvalidWaitTimeoutSeconds($command)) {
$deadlineFields = ConditionOccurrenceDeadline::fields($command);
if ($deadlineFields === null || self::hasInvalidWaitTimeoutSeconds($command)) {
return null;
}

Expand All @@ -6217,6 +6229,7 @@ private static function normalizeOpenConditionWaitCommand(array $command): ?arra
$command['condition_wait_occurrence_id'] ?? null,
),
'timeout_seconds' => self::normalizeWaitTimeoutSeconds($command),
...$deadlineFields,
...$parallelMetadata,
], static fn (mixed $value): bool => $value !== null);
}
Expand Down
5 changes: 5 additions & 0 deletions src/V2/Support/HistoryEventPayloadContract.php
Original file line number Diff line number Diff line change
Expand Up @@ -397,6 +397,8 @@ final class HistoryEventPayloadContract
'condition_definition_fingerprint',
'signal_wait_id',
'signal_name',
'deadline_policy',
'occurrence_deadline_at',
],
'TimerFired' => [
'reused_from_run_id',
Expand Down Expand Up @@ -586,6 +588,9 @@ final class HistoryEventPayloadContract
'parallel_group_size',
'parallel_group_index',
'parallel_group_path',
'deadline_policy',
'condition_wait_timeout_budget_seconds',
'occurrence_deadline_at',
],
'ConditionWaitSatisfied' => [
'condition_wait_id',
Expand Down
26 changes: 25 additions & 1 deletion src/V2/Support/WorkerProtocolVersion.php
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ final class WorkerProtocolVersion
* pagination semantics). Bump the minor for additive changes (new
* optional fields, new non-terminal command types).
*/
public const VERSION = '1.21';
public const VERSION = '1.22';

/**
* Worker registration capability for server-routed workflow query
Expand Down Expand Up @@ -677,6 +677,8 @@ public static function conditionWaitCommandShape(): array
'condition_definition_fingerprint',
'condition_wait_occurrence_id',
'timeout_seconds',
'deadline_policy',
'condition_wait_timeout_budget_seconds',
],
'condition_wait_occurrence_id' => [
'shape' => 'non-empty string',
Expand All @@ -685,6 +687,28 @@ public static function conditionWaitCommandShape(): array
'worker_capability' => self::CAPABILITY_CONDITION_WAIT_OCCURRENCE_IDENTITY,
'meaning' => 'explicit deterministic identity of one authored wait occurrence',
],
'deadline_policy' => [
'value' => 'occurrence',
'server_capability' => 'condition_wait_occurrence_deadline',
'negotiation' => 'opt in only when the runtime advertises the capability; keep recorded chains on their original policy',
'budget_field' => 'condition_wait_timeout_budget_seconds',
'budget_shape' => 'non-negative integer original timeout, at least timeout_seconds',
'required_with_policy' => [
'condition_wait_occurrence_id',
'timeout_seconds',
'condition_wait_timeout_budget_seconds',
],
'first_open' => 'timeout_seconds equals the original budget; the runtime sets A from its commit clock',
'reopen' => 'same run, occurrence, key, fingerprint and original budget reuse the first timer fire_at exactly',
'expired_anchor' => 'schedule due at A without firing inline or inventing a past dispatch time',
'refusal_reasons' => [
'condition_wait_occurrence_mismatch',
'condition_wait_occurrence_timed_out',
'condition_wait_anchor_unavailable',
],
'legacy_absence' => 'retain per-open scheduling and refuse a later opt-in to that chain',
'history_anchor_field' => 'occurrence_deadline_at',
],
'history' => [
'lifecycle_event_types' => [
'ConditionWaitOpened',
Expand Down
Loading
Loading