Skip to content

Commit ccc3b04

Browse files
rmcdanielrmcdaniel
authored
Merge pull request #831 from durable-workflow/release/2.5.30-replay-read-guards
release: qualify compatible replay decoding patch 2.5.30
2 parents 69203a7 + 2e46979 commit ccc3b04

7 files changed

Lines changed: 230 additions & 5 deletions

File tree

‎CHANGELOG.md‎

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

3+
## 2.5.30 - 2026-10-17
4+
5+
- Skip payload decoding for history event types that cannot match an activity
6+
or a parallel member failure during replay.
7+
- Preserve recorded activity retry policies, member identity and failure order.
8+
39
## 2.5.29 - 2026-10-17
410

511
- Reuse an event's decoded payload within each timer projection, avoiding

‎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.29",
82+
"product-train": "2.5.30",
8383
"laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
8484
"laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
8585
},

‎src/V2/Support/ActivityReplayIdentity.php‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,15 @@ public static function assertCall(
4242
$run->load('historyEvents');
4343
}
4444
foreach ($run->historyEvents as $event) {
45+
if ($event->event_type !== HistoryEventType::ActivityScheduled) {
46+
continue;
47+
}
48+
4549
$recordedSequence = $event->payload['sequence'] ?? null;
4650
if (is_string($recordedSequence) && is_numeric($recordedSequence)) {
4751
$recordedSequence = (int) $recordedSequence;
4852
}
49-
if ($event->event_type !== HistoryEventType::ActivityScheduled
50-
|| $recordedSequence !== $sequence
53+
if ($recordedSequence !== $sequence
5154
|| ($event->payload['execution_mode'] ?? null) === LocalActivityRuntime::EXECUTION_MODE
5255
|| ($event->payload['activity']['execution_mode'] ?? null) === LocalActivityRuntime::EXECUTION_MODE) {
5356
continue;

‎src/V2/Support/ParallelChildGroup.php‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1084,9 +1084,12 @@ private static function resolutionEventForMember(
10841084
$memberSequences,
10851085
$conditionResolutions
10861086
): bool {
1087+
if (! in_array($event->event_type, $types, true)) {
1088+
return false;
1089+
}
1090+
10871091
$sequence = $event->payload['sequence'] ?? null;
1088-
if (! in_array($event->event_type, $types, true)
1089-
|| ! is_int($sequence)
1092+
if (! is_int($sequence)
10901093
|| ! array_key_exists($sequence, $memberSequences)) {
10911094
return false;
10921095
}

‎tests/Fixtures/V2/ReplayRegression/replay-unrelated-payload-read-budget.json‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
{
2+
"$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
3+
"fixture_schema": "durable-workflow.replay-regression/v1",
4+
"id": "replay-unrelated-payload-read-budget",
5+
"protocol_version": "1.0",
6+
"bindings": ["php"],
7+
"workflow": {
8+
"type": "Tests\\Fixtures\\V2\\TestRedriveWorkflow",
9+
"arguments": ["Ada"],
10+
"payload_codec": "avro"
11+
},
12+
"history": [
13+
{
14+
"sequence": 1,
15+
"event_type": "WorkflowStarted",
16+
"payload": {},
17+
"recorded_at": "2026-10-17T12:00:00.000001Z"
18+
},
19+
{
20+
"sequence": 2,
21+
"event_type": "ActivityScheduled",
22+
"payload": {
23+
"sequence": 1,
24+
"activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity",
25+
"activity_execution_id": "01j00000000000000000000001"
26+
},
27+
"recorded_at": "2026-10-17T12:00:01.000001Z"
28+
},
29+
{
30+
"sequence": 3,
31+
"event_type": "ActivityCompleted",
32+
"payload": {
33+
"sequence": 1,
34+
"activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity",
35+
"activity_execution_id": "01j00000000000000000000001",
36+
"result": "wwHioz3/VYAiNwoWSGVsbG8sIEFkYSE=",
37+
"payload_codec": "avro"
38+
},
39+
"recorded_at": "2026-10-17T12:00:02.000001Z"
40+
}
41+
],
42+
"expected": {
43+
"completed": false,
44+
"result": null,
45+
"commands": [
46+
{
47+
"type": "schedule_activity",
48+
"activity_type": "Tests\\Fixtures\\V2\\TestRedriveSecondActivity",
49+
"payload_codec": "avro"
50+
}
51+
]
52+
}
53+
}

‎tests/Unit/V2/ReplayPayloadReadGuardTest.php‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Tests\Unit\V2;
6+
7+
use Illuminate\Database\Eloquent\Collection;
8+
use PHPUnit\Framework\TestCase;
9+
use Workflow\V2\Enums\HistoryEventType;
10+
use Workflow\V2\Models\WorkflowHistoryEvent;
11+
use Workflow\V2\Models\WorkflowRun;
12+
use Workflow\V2\Support\ActivityCall;
13+
use Workflow\V2\Support\ActivityReplayIdentity;
14+
use Workflow\V2\Support\ParallelChildGroup;
15+
16+
final class ReplayPayloadReadGuardTest extends TestCase
17+
{
18+
public function testActivityIdentityDoesNotDecodeUnrelatedEventTypes(): void
19+
{
20+
$unrelated = self::event(HistoryEventType::WorkflowStarted, 1, 7);
21+
$scheduled = self::event(HistoryEventType::ActivityScheduled, 2, 7);
22+
$after = self::event(HistoryEventType::ActivityCompleted, 3, 7);
23+
$events = new Collection([$unrelated, $scheduled, $after]);
24+
$before = $events->map->getAttributes()
25+
->all();
26+
$run = new WorkflowRun();
27+
$run->setRelation('historyEvents', $events);
28+
29+
ActivityReplayIdentity::assertCall($run, new ActivityCall('ExampleActivity', [1]), 7);
30+
31+
self::assertSame(0, $unrelated->payloadReads);
32+
self::assertGreaterThan(0, $scheduled->payloadReads);
33+
self::assertSame(0, $after->payloadReads);
34+
self::assertSame($before, $events->map->getAttributes()->all());
35+
}
36+
37+
public function testFailureLookupDoesNotDecodeUnrelatedEventTypes(): void
38+
{
39+
$events = new Collection();
40+
foreach ([
41+
HistoryEventType::WorkflowStarted,
42+
HistoryEventType::ActivityScheduled,
43+
HistoryEventType::ActivityStarted,
44+
HistoryEventType::ActivityCompleted,
45+
HistoryEventType::TimerScheduled,
46+
HistoryEventType::TimerFired,
47+
HistoryEventType::WorkflowCompleted,
48+
] as $index => $type) {
49+
$events->push(self::event($type, $index + 1, 1));
50+
}
51+
$before = $events->map->getAttributes()
52+
->all();
53+
$run = new WorkflowRun();
54+
$run->setRelation('historyEvents', $events);
55+
56+
self::assertNull(ParallelChildGroup::memberFailureResolution($run, 1, 3));
57+
58+
foreach ($events as $event) {
59+
self::assertSame(0, $event->payloadReads);
60+
}
61+
self::assertSame($before, $events->map->getAttributes()->all());
62+
}
63+
64+
public function testFailureLookupKeepsOriginalMemberAndHistoryOrdering(): void
65+
{
66+
$unrelated = self::event(HistoryEventType::ActivityCompleted, 1, 1);
67+
$otherGroup = self::event(HistoryEventType::ActivityFailed, 2, 9);
68+
$later = self::event(HistoryEventType::ActivityFailed, 40, 2);
69+
$earlier = self::event(HistoryEventType::ActivityTimedOut, 30, 3);
70+
$events = new Collection([$unrelated, $otherGroup, $later, $earlier]);
71+
$before = $events->map->getAttributes()
72+
->all();
73+
$run = new WorkflowRun();
74+
$run->setRelation('historyEvents', $events);
75+
76+
self::assertSame($earlier, ParallelChildGroup::memberFailureResolution($run, 1, 3));
77+
self::assertGreaterThan(0, $earlier->payloadReads);
78+
self::assertSame($before, $events->map->getAttributes()->all());
79+
}
80+
81+
private static function event(
82+
HistoryEventType $type,
83+
int $historySequence,
84+
int $commandSequence
85+
): ReadCountHistoryEvent {
86+
return new ReadCountHistoryEvent([
87+
'id' => 'event-' . $historySequence,
88+
'sequence' => $historySequence,
89+
'event_type' => $type,
90+
'payload' => [
91+
'sequence' => $commandSequence,
92+
],
93+
]);
94+
}
95+
}
96+
97+
final class ReadCountHistoryEvent extends WorkflowHistoryEvent
98+
{
99+
public int $payloadReads = 0;
100+
101+
public function getAttribute($key): mixed
102+
{
103+
if ($key === 'payload') {
104+
++$this->payloadReads;
105+
}
106+
107+
return parent::getAttribute($key);
108+
}
109+
}

‎tests/Unit/V2/ReplayRegressionCorpusTest.php‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,16 @@ public function testFixtureExecutesThroughColdReplayRunner(array $fixture): void
113113
$this->fail("{$fixture['id']} did not reject its malformed replay history.");
114114
}
115115

116+
$profiledWorkflow = null;
117+
if ($fixture['id'] === 'replay-unrelated-payload-read-budget') {
118+
Container::getInstance()->resolving(
119+
$workflowClass,
120+
static function (Workflow $instance) use (&$profiledWorkflow): void {
121+
$profiledWorkflow = $instance;
122+
},
123+
);
124+
}
125+
116126
$runner = WorkflowFiberRunner::forClass(
117127
$workflowClass,
118128
'regression-corpus-' . $fixture['id'],
@@ -122,8 +132,35 @@ public function testFixtureExecutesThroughColdReplayRunner(array $fixture): void
122132
$fixture['history'] ?? [],
123133
);
124134

135+
if ($fixture['id'] === 'replay-unrelated-payload-read-budget') {
136+
$this->assertInstanceOf(Workflow::class, $profiledWorkflow);
137+
$events = $profiledWorkflow->run->historyEvents->map(
138+
static function (WorkflowHistoryEvent $event): ReplayBudgetHistoryEvent {
139+
$profiled = new ReplayBudgetHistoryEvent();
140+
$profiled->setConnection($event->getConnectionName());
141+
$profiled->setTable($event->getTable());
142+
$profiled->setRawAttributes($event->getAttributes());
143+
$profiled->setRelations($event->getRelations());
144+
$profiled->exists = $event->exists;
145+
$profiled->wasRecentlyCreated = $event->wasRecentlyCreated;
146+
147+
return $profiled;
148+
},
149+
);
150+
$profiledWorkflow->run->setRelation('historyEvents', $events);
151+
ReplayBudgetHistoryEvent::$startedPayloadReads = 0;
152+
}
153+
125154
$step = $runner->step();
126155

156+
if ($fixture['id'] === 'replay-unrelated-payload-read-budget') {
157+
$this->assertSame(
158+
0,
159+
ReplayBudgetHistoryEvent::$startedPayloadReads,
160+
'Activity lookups must not decode start metadata already consumed during bridge initialization.',
161+
);
162+
}
163+
127164
if (isset($fixture['command_sequence'])) {
128165
foreach ($fixture['command_sequence'] as $index => $expectedStep) {
129166
$this->assertStepMatches($expectedStep, $step, "{$fixture['id']} command step {$index}");
@@ -469,3 +506,17 @@ private function assertArrayContains(array $expected, array $actual, string $con
469506
}
470507
}
471508
}
509+
510+
final class ReplayBudgetHistoryEvent extends WorkflowHistoryEvent
511+
{
512+
public static int $startedPayloadReads = 0;
513+
514+
public function getAttribute($key): mixed
515+
{
516+
if ($key === 'payload' && parent::getAttribute('event_type') === HistoryEventType::WorkflowStarted) {
517+
++self::$startedPayloadReads;
518+
}
519+
520+
return parent::getAttribute($key);
521+
}
522+
}

0 commit comments

Comments
 (0)