Skip to content

Commit c677f37

Browse files
rmcdanielrmcdaniel
authored
Backport explicit activity retry replay fences to stable 2.5.16 (#777)
* Refuse changed explicit activity retry policy during embedded replay (#776) Refuse changed explicit activity retry policy during embedded replay * Prepare compatible 2.5.16 retry replay release --------- Co-authored-by: Durable Workflow <support@durable-workflow.com>
1 parent 90e3ac9 commit c677f37

14 files changed

Lines changed: 724 additions & 1 deletion

‎CHANGELOG.md‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,14 @@
11
# Changelog
22

3+
## 2.5.16 - 2026-10-13
4+
5+
- Refuse changed explicitly authored activity retry policies during embedded
6+
replay before consuming an original result or scheduling new work. Compare
7+
complete recorded backoff lists, including unused entries.
8+
- Preserve the original run and task for matching-code repair. Apply the check
9+
to direct, grouped, selected and query replay while retaining class-derived
10+
retry defaults and legacy histories without a recorded policy.
11+
312
## 2.5.15 - 2026-10-13
413

514
- Preserve unclaimed database-queue jobs when SQLite refuses a reservation write

‎composer.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@
7979
"dev-main": "2.0.x-dev"
8080
},
8181
"durable-workflow": {
82-
"product-train": "2.5.15",
82+
"product-train": "2.5.16",
8383
"laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
8484
"laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
8585
},

‎scripts/ci/validate-regression-corpus.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1280,6 +1280,7 @@ def _replay_expected_failure(value: Any, context: str) -> Mapping[str, str]:
12801280
WORKFLOW_HISTORY_SHAPE_MISMATCH,
12811281
VERSION_MARKER_INCOMPATIBLE_RANGE,
12821282
VERSION_MARKER_HISTORY_RANGE_INVALID,
1283+
"activity_retry_policy_mismatch",
12831284
UNSUPPORTED_PAYLOAD_CODEC,
12841285
}:
12851286
raise CorpusError(f"{context}.type is unsupported")

‎src/V2/Exceptions/ActivityRetryPolicyMismatchException.php‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Workflow\V2\Exceptions;
6+
7+
use LogicException;
8+
9+
final class ActivityRetryPolicyMismatchException extends LogicException
10+
{
11+
public function __construct(
12+
public readonly int $workflowSequence,
13+
public readonly string $policyField,
14+
) {
15+
parent::__construct(sprintf(
16+
'Activity retry policy at workflow sequence %d differs in [%s]. Restore the recorded per-call policy or use a versioned workflow change before repairing this run.',
17+
$workflowSequence,
18+
$policyField,
19+
));
20+
}
21+
}

‎src/V2/Support/ActivityReplayIdentity.php‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Workflow\V2\Support;
6+
7+
use Workflow\V2\Enums\HistoryEventType;
8+
use Workflow\V2\Exceptions\ActivityRetryPolicyMismatchException;
9+
use Workflow\V2\Models\WorkflowRun;
10+
11+
final class ActivityReplayIdentity
12+
{
13+
/**
14+
* @param callable(int): int|null $historySequence
15+
*/
16+
public static function assertCall(
17+
WorkflowRun $run,
18+
mixed $call,
19+
int $sequence,
20+
?callable $historySequence = null,
21+
): void {
22+
if ($call instanceof DurableOperationHandle) {
23+
self::assertCall($run, $call->call, $call->baseSequence);
24+
25+
return;
26+
}
27+
28+
if ($call instanceof AllCall) {
29+
foreach ($call->leafDescriptors($sequence) as $descriptor) {
30+
self::assertCall($run, $descriptor['call'], $sequence + $descriptor['offset'], $historySequence);
31+
}
32+
33+
return;
34+
}
35+
36+
if (! $call instanceof ActivityCall) {
37+
return;
38+
}
39+
40+
$sequence = $historySequence === null ? $sequence : $historySequence($sequence);
41+
if (! $run->relationLoaded('historyEvents')) {
42+
$run->load('historyEvents');
43+
}
44+
foreach ($run->historyEvents as $event) {
45+
$recordedSequence = $event->payload['sequence'] ?? null;
46+
if (is_string($recordedSequence) && is_numeric($recordedSequence)) {
47+
$recordedSequence = (int) $recordedSequence;
48+
}
49+
if ($event->event_type !== HistoryEventType::ActivityScheduled
50+
|| $recordedSequence !== $sequence
51+
|| ($event->payload['execution_mode'] ?? null) === LocalActivityRuntime::EXECUTION_MODE
52+
|| ($event->payload['activity']['execution_mode'] ?? null) === LocalActivityRuntime::EXECUTION_MODE) {
53+
continue;
54+
}
55+
56+
$snapshot = $event->payload['activity']['retry_policy'] ?? $event->payload['retry_policy'] ?? null;
57+
if ($snapshot === null) {
58+
// Legacy absence is not an invitation to invent today's class defaults.
59+
return;
60+
}
61+
if (! is_array($snapshot)) {
62+
throw new ActivityRetryPolicyMismatchException($sequence, 'retry_policy');
63+
}
64+
65+
self::assertPolicy($snapshot, $call->options, $sequence);
66+
67+
return;
68+
}
69+
}
70+
71+
/**
72+
* Compare explicit authored overrides only; never instantiate an activity
73+
* or recompute its class-derived budget during replay.
74+
*
75+
* @param array<string, mixed> $snapshot
76+
*/
77+
public static function assertPolicy(array $snapshot, ?ActivityOptions $options, int $sequence): void
78+
{
79+
$authored = [
80+
'max_attempts' => $options?->maxAttempts,
81+
'backoff_seconds' => $options?->backoff,
82+
'start_to_close_timeout' => $options?->startToCloseTimeout,
83+
'schedule_to_start_timeout' => $options?->scheduleToStartTimeout,
84+
'schedule_to_close_timeout' => $options?->scheduleToCloseTimeout,
85+
'heartbeat_timeout' => $options?->heartbeatTimeout,
86+
'non_retryable_error_types' => $options?->nonRetryableErrorTypes ?: null,
87+
];
88+
89+
$normalized = ActivityRetryPolicy::snapshotExternal(null, $options) ?? [
90+
'non_retryable_error_types' => [],
91+
];
92+
foreach ($snapshot as $field => $recorded) {
93+
$valid = match ($field) {
94+
'snapshot_version' => $recorded === ActivityRetryPolicy::SNAPSHOT_VERSION,
95+
'max_attempts' => $recorded === null || (is_int($recorded) && $recorded > 0),
96+
'backoff_seconds' => is_array($recorded) && array_is_list($recorded)
97+
&& array_filter($recorded, static fn (mixed $value): bool => ! is_int($value) || $value < 0) === [],
98+
'non_retryable_error_types' => is_array($recorded) && array_is_list($recorded)
99+
&& array_filter(
100+
$recorded,
101+
static fn (mixed $value): bool => ! is_string($value) || trim($value) === ''
102+
) === [],
103+
'start_to_close_timeout', 'schedule_to_start_timeout', 'schedule_to_close_timeout', 'heartbeat_timeout'
104+
=> $recorded === null || (is_int($recorded) && $recorded >= 0),
105+
default => true,
106+
};
107+
if (! $valid) {
108+
throw new ActivityRetryPolicyMismatchException($sequence, $field);
109+
}
110+
111+
$current = $authored[$field] ?? null;
112+
if (in_array($field, ['max_attempts', 'backoff_seconds'], true) && $current === null) {
113+
// These two fields can originate from activity-class defaults.
114+
continue;
115+
}
116+
if (! array_key_exists($field, $authored)) {
117+
continue;
118+
}
119+
// Apply exactly the same normalization as initial scheduling.
120+
if ($recorded !== ($normalized[$field] ?? null)) {
121+
throw new ActivityRetryPolicyMismatchException($sequence, $field);
122+
}
123+
}
124+
}
125+
}

‎src/V2/Support/QueryStateReplayer.php‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,16 @@ public function replayState(WorkflowRun $run): ReplayState
8686
return new ReplayState($workflow, $sequence, null);
8787
}
8888

89+
ActivityReplayIdentity::assertCall(
90+
$run,
91+
$current,
92+
$sequence,
93+
fn (int $position): int => $this->historySequenceForReplayPosition(
94+
$historySequencesByPosition,
95+
$position
96+
),
97+
);
98+
8999
if ($current instanceof CancelDurableOperationCall) {
90100
$cancelled = $this->selectionOperationCancelledEvent($run, $current->handle);
91101
if (! $cancelled instanceof WorkflowHistoryEvent && ! $this->unmarkedCancellationWasCommitted(

‎src/V2/Support/WorkflowExecutor.php‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
use Workflow\V2\Enums\TaskType;
3333
use Workflow\V2\Enums\TimerStatus;
3434
use Workflow\V2\Enums\UpdateStatus;
35+
use Workflow\V2\Exceptions\ActivityRetryPolicyMismatchException;
3536
use Workflow\V2\Exceptions\ConditionWaitDefinitionMismatchException;
3637
use Workflow\V2\Exceptions\DurableOperationCancelledException;
3738
use Workflow\V2\Exceptions\HistoryEventShapeMismatchException;
@@ -111,6 +112,7 @@ public function run(WorkflowRun $run, WorkflowTask $task): ?WorkflowTask
111112
++$sequence;
112113
$this->syncWorkflowCursor($workflow, $sequence);
113114
}
115+
ActivityReplayIdentity::assertCall($run, $current, $sequence);
114116
} catch (Throwable $throwable) {
115117
$this->failRun($run, $task, $throwable, 'workflow_run', $run->id);
116118

@@ -4329,6 +4331,25 @@ private function failRun(
43294331
return;
43304332
}
43314333

4334+
if ($throwable instanceof ActivityRetryPolicyMismatchException) {
4335+
$payload = is_array($task->payload) ? $task->payload : [];
4336+
$payload['replay_blocked'] = true;
4337+
$payload['replay_blocked_reason'] = 'activity_retry_policy_mismatch';
4338+
$payload['replay_blocked_workflow_sequence'] = $throwable->workflowSequence;
4339+
$payload['replay_blocked_policy_field'] = $throwable->policyField;
4340+
$task->forceFill([
4341+
'status' => TaskStatus::Failed,
4342+
'payload' => $payload,
4343+
'last_error' => $throwable->getMessage(),
4344+
'lease_expires_at' => null,
4345+
])->save();
4346+
$this->projectRun(
4347+
$run->fresh(['instance', 'tasks', 'activityExecutions', 'timers', 'failures', 'historyEvents'])
4348+
);
4349+
4350+
return;
4351+
}
4352+
43324353
if ($throwable instanceof ConditionWaitDefinitionMismatchException) {
43334354
$this->blockReplayUntilCompatibleConditionWaitDefinition($run, $task, $throwable);
43344355

‎src/V2/Support/WorkflowFiberRunner.php‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -319,6 +319,12 @@ private function nextObservableStep(): WorkflowStep
319319
));
320320
}
321321

322+
ActivityReplayIdentity::assertCall(
323+
$this->workflow->run,
324+
$current,
325+
$this->historySequenceForCurrentPosition() ?? $this->sequence,
326+
);
327+
322328
$this->applyRecordedUpdatesForCurrentPosition();
323329

324330
$historySequence = $this->historySequenceForCurrentPosition();

0 commit comments

Comments
 (0)