From 532e7150c46d5d08b8981b675260a445ca2777fe Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 7 Oct 2026 08:53:59 +0000 Subject: [PATCH 1/2] fix: preserve legacy version fallback in offline replay --- scripts/ci/validate-regression-corpus.py | 1 + src/V2/Support/QueryStateReplayer.php | 12 +- .../V2EmbeddedReplayRegressionCorpusTest.php | 48 ++++ .../legacy-version-shared-step-position.json | 75 +++++++ .../V2/TestLegacyPatchReplayWorkflow.php | 46 ++++ .../V2/LegacyVersionOfflineReplayTest.php | 207 ++++++++++++++++++ 6 files changed, 385 insertions(+), 4 deletions(-) create mode 100644 tests/Fixtures/V2/ReplayRegression/legacy-version-shared-step-position.json create mode 100644 tests/Fixtures/V2/TestLegacyPatchReplayWorkflow.php create mode 100644 tests/Unit/V2/LegacyVersionOfflineReplayTest.php diff --git a/scripts/ci/validate-regression-corpus.py b/scripts/ci/validate-regression-corpus.py index 8f6310240..d59b171ef 100644 --- a/scripts/ci/validate-regression-corpus.py +++ b/scripts/ci/validate-regression-corpus.py @@ -161,6 +161,7 @@ "tests/Fixtures/V2/TestParallelChildReplayWorkflow.php", "tests/Fixtures/V2/TestSequentialChildReplayWorkflow.php", "tests/Fixtures/V2/TestServiceResponseReplayWorkflow.php", + "tests/Fixtures/V2/TestLegacyPatchReplayWorkflow.php", "tests/Fixtures/V2/TestServiceGroupedConditionReopenWorkflow.php", "tests/Fixtures/V2/TestSignalResumedParallelWorkflow.php", "tests/Fixtures/V2/TestCancellationContextWorkflow.php", diff --git a/src/V2/Support/QueryStateReplayer.php b/src/V2/Support/QueryStateReplayer.php index 81ecfea0d..04c42393d 100644 --- a/src/V2/Support/QueryStateReplayer.php +++ b/src/V2/Support/QueryStateReplayer.php @@ -326,13 +326,17 @@ public function replayState(WorkflowRun $run): ReplayState $historySequence = $this->historySequenceForReplayPosition($historySequencesByPosition, $sequence); $this->applyRecordedUpdates($run, $workflow, $historySequence); - WorkflowStepHistory::assertCompatible($run, $historySequence, WorkflowStepHistory::VERSION_MARKER, [ - 'change_id' => $current->changeId, - ]); - $versionEvent = $this->versionMarkerEvent($run, $historySequence); $resolution = VersionResolver::resolve($run, $versionEvent, $current, $historySequence); + // A legacy default occupies no history slot. The next durable + // step can already be recorded at this position. + if ($resolution->advancesSequence) { + WorkflowStepHistory::assertCompatible($run, $historySequence, WorkflowStepHistory::VERSION_MARKER, [ + 'change_id' => $current->changeId, + ]); + } + $this->syncWorkflowCursor($workflow, $sequence + ($resolution->advancesSequence ? 1 : 0)); $current = $workflowExecution->send( $current->resolveValue($resolution->version), diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 4955ddc1f..954242bf4 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -8,6 +8,7 @@ use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; use Symfony\Component\Uid\Ulid; +use Tests\Fixtures\V2\TestQueryReplayGuardActivity; use Tests\TestCase; use Throwable; use Workflow\Serializers\Serializer; @@ -39,6 +40,7 @@ use Workflow\V2\Support\EmbeddedV2HistoryImport; use Workflow\V2\Support\HistoryExport; use Workflow\V2\Support\QueryStateReplayer; +use Workflow\V2\Support\ReplayDiff; use Workflow\V2\Support\RunActivityView; use Workflow\V2\Support\RunTimelineProjector; use Workflow\V2\Support\WorkflowFiberRunner; @@ -121,6 +123,12 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void $this->assertRunInheritedScopedCleanupAfterColdReload($fixture); } + if ($fixture['id'] === 'legacy-version-shared-step-position') { + $this->assertLegacyVersionSharesTheRecordedStepPosition($fixture); + + continue; + } + if ($fixture['id'] === 'cooperative-delivery-before-wait-sequence') { $run = $this->createRunFromFixture($fixture); $this->assertSame(2, WorkflowStepHistory::nextDurableCommandSequence($run->fresh())); @@ -1023,6 +1031,46 @@ private function clearWorkflowState(): void } } + /** + * @param array $fixture + */ + private function assertLegacyVersionSharesTheRecordedStepPosition(array $fixture): void + { + $run = $this->createRunFromFixture($fixture); + $started = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::WorkflowStarted->value) + ->firstOrFail(); + $payload = $started->payload; + unset($payload['workflow_definition_fingerprint']); + $started->forceFill([ + 'payload' => $payload, + ])->save(); + $bundle = HistoryExport::forRun($run->fresh()); + $before = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->get()->toArray(); + TestQueryReplayGuardActivity::$executions = 0; + + $this->assertSame( + $fixture['expected']['result'], + (new QueryStateReplayer())->query($run->fresh(), 'currentState'), + ); + $this->assertSame(ReplayDiff::STATUS_REPLAYED, (new ReplayDiff())->diffExport($bundle)['status']); + $this->assertSame( + $before, + WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)->get()->toArray(), + ); + + $this->runReadyWorkflowTask($run); + $workflow = WorkflowStub::loadRun($run->id); + $this->assertTrue($workflow->completed()); + $this->assertSame($fixture['expected']['result'], $workflow->output()); + $this->assertSame(0, TestQueryReplayGuardActivity::$executions); + $this->assertFalse(WorkflowHistoryEvent::query() + ->where('workflow_run_id', $run->id) + ->where('event_type', HistoryEventType::VersionMarkerRecorded->value) + ->exists()); + } + /** * @param array $fixture */ diff --git a/tests/Fixtures/V2/ReplayRegression/legacy-version-shared-step-position.json b/tests/Fixtures/V2/ReplayRegression/legacy-version-shared-step-position.json new file mode 100644 index 000000000..775465224 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/legacy-version-shared-step-position.json @@ -0,0 +1,75 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "legacy-version-shared-step-position", + "protocol_version": "1.19", + "bindings": ["php"], + "consumers": ["workflow-fiber-runner", "query-state-replayer", "workflow-executor"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestLegacyPatchReplayWorkflow", + "arguments": [], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-10-01T12:00:00+00:00" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestQueryReplayGuardActivity", + "activity_execution_id": "01j00000000000000000000001" + }, + "recorded_at": "2026-10-01T12:00:01+00:00" + }, + { + "sequence": 3, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "Tests\\Fixtures\\V2\\TestQueryReplayGuardActivity", + "activity_execution_id": "01j00000000000000000000001", + "result": "wwHioz3/VYAiNwoecmVjb3JkZWQtcmVzdWx0", + "payload_codec": "avro" + }, + "recorded_at": "2026-10-01T12:00:02+00:00" + }, + { + "sequence": 4, + "event_type": "TimerScheduled", + "payload": { + "sequence": 2, + "timer_id": "01j00000000000000000000002", + "timer_kind": "durable_timer", + "delay_seconds": 60, + "fire_at": "2026-10-01T12:01:02+00:00" + }, + "recorded_at": "2026-10-01T12:00:02+00:00" + }, + { + "sequence": 5, + "event_type": "TimerFired", + "payload": { + "sequence": 2, + "timer_id": "01j00000000000000000000002", + "timer_kind": "durable_timer", + "fired_at": "2026-10-01T12:01:02+00:00" + }, + "recorded_at": "2026-10-01T12:01:02+00:00" + } + ], + "expected": { + "completed": true, + "result": { + "patched": false, + "value": "recorded-result", + "stage": "completed" + }, + "commands": [{"type": "complete_workflow"}] + } +} diff --git a/tests/Fixtures/V2/TestLegacyPatchReplayWorkflow.php b/tests/Fixtures/V2/TestLegacyPatchReplayWorkflow.php new file mode 100644 index 000000000..3930573f0 --- /dev/null +++ b/tests/Fixtures/V2/TestLegacyPatchReplayWorkflow.php @@ -0,0 +1,46 @@ +patched = Workflow::patched('legacy-upgrade'); + } + + $this->value = Workflow::activity(TestQueryReplayGuardActivity::class); + + if ($patchBefore === 'timer') { + $this->patched = Workflow::patched('legacy-upgrade'); + } + + $this->stage = 'waiting-for-timer'; + Workflow::timer(60); + $this->stage = 'completed'; + + return $this->currentState(); + } + + #[QueryMethod] + public function currentState(): array + { + return [ + 'patched' => $this->patched, + 'value' => $this->value, + 'stage' => $this->stage, + ]; + } +} diff --git a/tests/Unit/V2/LegacyVersionOfflineReplayTest.php b/tests/Unit/V2/LegacyVersionOfflineReplayTest.php new file mode 100644 index 000000000..00c848176 --- /dev/null +++ b/tests/Unit/V2/LegacyVersionOfflineReplayTest.php @@ -0,0 +1,207 @@ +bundle($patchBefore, $completed); + if ($legacyEvidence === 'changed-fingerprint') { + $bundle['history_events'][0]['payload']['workflow_definition_fingerprint'] = + 'sha256:' . hash('sha256', 'definition-before-patched-call'); + } elseif ($legacyEvidence === 'older-compatibility') { + $bundle['workflow']['compatibility'] = 'build-before-patch'; + $bundle['history_events'][0]['payload']['workflow_definition_fingerprint'] = + WorkflowDefinition::fingerprint(TestLegacyPatchReplayWorkflow::class); + config([ + 'workflows.v2.compatibility.current' => 'build-after-patch', + ]); + } + $before = $bundle; + $expected = [ + 'patched' => false, + 'value' => 'recorded-result', + 'stage' => $completed ? 'completed' : 'waiting-for-timer', + ]; + TestQueryReplayGuardActivity::$executions = 0; + + foreach ([1, 2] as $repetition) { + $replayer = new WorkflowReplayer(); + $run = $replayer->runFromHistoryExport($bundle); + $state = $replayer->replay($run); + $this->assertSame($expected, $state->workflow->currentState()); + $this->assertSame($completed ? 3 : 2, $state->sequence); + if ($completed) { + $this->assertNull($state->current); + } else { + $this->assertInstanceOf(TimerCall::class, $state->current); + } + $this->assertSame($expected, (new QueryStateReplayer())->query($run, 'currentState')); + $report = (new ReplayDiff())->diffExport($bundle); + $this->assertSame(ReplayDiff::STATUS_REPLAYED, $report['status'], json_encode($report)); + $this->assertSame(ReplayDiff::REASON_NONE, $report['reason']); + $this->assertSame(0, TestQueryReplayGuardActivity::$executions); + $this->assertSame($before, $bundle); + } + } + + public static function legacyHistories(): array + { + $cases = []; + foreach (['activity', 'timer'] as $position) { + foreach ([false, true] as $completed) { + foreach (['missing-fingerprint', 'changed-fingerprint', 'older-compatibility'] as $evidence) { + $cases[$position . '-' . ($completed ? 'completed' : 'waiting') . '-' . $evidence] = [ + $position, $completed, $evidence, + ]; + } + } + } + + return $cases; + } + + #[DataProvider('invalidHistories')] + public function testFallbackDoesNotHideMalformedMarkersOrStepDrift(string $case): void + { + $bundle = $this->bundle('activity', true); + $reason = ReplayDiff::REASON_REPLAY_ERROR; + if ($case === 'current-definition') { + $bundle['history_events'][0]['payload']['workflow_definition_fingerprint'] = + WorkflowDefinition::fingerprint(TestLegacyPatchReplayWorkflow::class); + $reason = ReplayDiff::REASON_SHAPE_MISMATCH; + } elseif ($case === 'activity-drift') { + foreach ([1, 2] as $index) { + $bundle['history_events'][$index]['payload']['activity_type'] = 'ChangedActivity'; + } + $reason = ReplayDiff::REASON_SHAPE_MISMATCH; + } elseif ($case === 'timer-drift') { + $bundle = $this->bundle('timer', true); + foreach ([ + 3 => 'ActivityScheduled', + 4 => 'ActivityCompleted', + ] as $index => $type) { + $bundle['history_events'][$index]['type'] = $type; + $bundle['history_events'][$index]['payload']['activity_type'] = TestQueryReplayGuardActivity::class; + } + $reason = ReplayDiff::REASON_SHAPE_MISMATCH; + } else { + $payload = [ + 'sequence' => 1, + 'change_id' => 'legacy-upgrade', + 'version' => 1, + ]; + if ($case === 'missing-change-id') { + unset($payload['change_id']); + } elseif ($case === 'wrong-change-id') { + $payload['change_id'] = 'other-change'; + } elseif ($case === 'non-integer-version') { + $payload['version'] = '1'; + } elseif ($case === 'unsupported-version') { + $payload['version'] = 2; + } elseif ($case === 'marker-and-activity') { + $reason = ReplayDiff::REASON_SHAPE_MISMATCH; + } + if ($case !== 'marker-and-activity') { + foreach ($bundle['history_events'] as &$event) { + if (isset($event['payload']['sequence'])) { + ++$event['payload']['sequence']; + } + } + unset($event); + } + array_splice($bundle['history_events'], 1, 0, [[ + 'id' => 'marker-1', + 'sequence' => 2, + 'type' => 'VersionMarkerRecorded', + 'payload' => $payload, + 'recorded_at' => '2026-10-01T12:00:01+00:00', + ]]); + foreach ($bundle['history_events'] as $index => &$event) { + $event['sequence'] = $index + 1; + } + unset($event); + $bundle['workflow']['last_history_sequence'] = count($bundle['history_events']); + } + $before = $bundle; + $report = (new ReplayDiff())->diffExport($bundle); + + $this->assertSame($reason, $report['reason'], json_encode($report)); + $this->assertSame( + $reason === ReplayDiff::REASON_SHAPE_MISMATCH ? ReplayDiff::STATUS_DRIFTED : ReplayDiff::STATUS_FAILED, + $report['status'], + ); + $this->assertSame($before, $bundle); + } + + public static function invalidHistories(): array + { + return array_combine( + $cases = [ + 'current-definition', 'activity-drift', 'timer-drift', 'missing-change-id', + 'wrong-change-id', 'non-integer-version', 'unsupported-version', 'marker-and-activity', + ], + array_map(static fn (string $case): array => [$case], $cases), + ); + } + + private function bundle(string $patchBefore, bool $completed): array + { + $fixture = json_decode(file_get_contents( + __DIR__ . '/../../Fixtures/V2/ReplayRegression/legacy-version-shared-step-position.json' + ), true, flags: JSON_THROW_ON_ERROR); + $events = $fixture['history']; + if (! $completed) { + array_pop($events); + } + + return [ + 'schema' => HistoryExport::SCHEMA, + 'schema_version' => HistoryExport::SCHEMA_VERSION, + 'workflow' => [ + 'instance_id' => 'legacy-instance', + 'run_id' => 'legacy-run', + 'workflow_type' => $fixture['workflow']['type'], + 'workflow_class' => $fixture['workflow']['type'], + 'status' => $completed ? 'completed' : 'waiting', + 'last_history_sequence' => count($events), + ], + 'payloads' => [ + 'codec' => 'avro', + 'arguments' => [ + 'available' => true, + 'data' => Serializer::serializeWithCodec('avro', [$patchBefore]), + ], + 'output' => [ + 'available' => false, + ], + ], + 'history_events' => array_map(static fn (array $event): array => [ + 'id' => 'event-' . $event['sequence'], + 'sequence' => $event['sequence'], + 'type' => $event['event_type'], + 'payload' => $event['payload'], + 'recorded_at' => $event['recorded_at'], + ], $events), + ]; + } +} From 5ebda8e8990af30f7a32d82ce503695eb0b28a01 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 7 Oct 2026 08:54:58 +0000 Subject: [PATCH 2/2] chore: prepare Workflow 2.5.1 patch release --- composer.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/composer.json b/composer.json index f3577c54e..d1ba37b15 100644 --- a/composer.json +++ b/composer.json @@ -79,7 +79,7 @@ "dev-main": "2.0.x-dev" }, "durable-workflow": { - "product-train": "2.5.0", + "product-train": "2.5.1", "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json", "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json" },