diff --git a/CHANGELOG.md b/CHANGELOG.md index a393fdaa3..30139da83 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,14 @@ registered for its type later. Failures recorded by earlier releases have no boundary and still refuse with `unrecorded_failure_boundary` (#864). +- Redrive refuses a run a service worker started when its completed prefix + has a memo upsert, search-attribute upsert or signal wait to reuse, with + `service_redrive_history_not_qualified`, before writing anything. Embedded + runs, and service runs whose prefix holds only activities, side effects, + version markers and fired timers, redrive as before. The refusal lifts only + through a replay-capability check once SDK replay of those reused events + ships (durable-workflow/sdk-python#235, durable-workflow/sdk-php#244, + durable-workflow/sdk-rust#172) (#864). ## 2.18.0 - 2026-11-03 diff --git a/src/V2/Support/FailedRunRedriveEligibility.php b/src/V2/Support/FailedRunRedriveEligibility.php index 77d7657c2..9ff5d57c3 100644 --- a/src/V2/Support/FailedRunRedriveEligibility.php +++ b/src/V2/Support/FailedRunRedriveEligibility.php @@ -107,6 +107,17 @@ public static function evaluate( return self::refuse('unsupported_source_metadata'); } + // Service-mode SDK workers are qualified to replay reused activities, + // side effects, version markers and timers. Reused signal, memo and + // search-attribute steps are not qualified yet, so a run a service + // worker started refuses them before anything is written. + if ( + ($startedPayload['workflow_definition_fingerprint_source'] ?? null) === 'worker' + && self::reusesServiceUnqualifiedSteps($plan['reused_steps']) + ) { + return self::refuse('service_redrive_history_not_qualified'); + } + $sourceFingerprint = $startedPayload['workflow_definition_fingerprint'] ?? null; $definitionChange = null; @@ -190,6 +201,27 @@ public static function blockedReason(WorkflowRun $run, ?WorkflowRun $currentRun) return self::evaluate($run, $currentRun, fingerprintPresentedAtRedrive: true)['reason']; } + /** + * @param list}> $reusedSteps + */ + private static function reusesServiceUnqualifiedSteps(array $reusedSteps): bool + { + foreach ($reusedSteps as $step) { + if ( + isset($step['signal_events']) + || in_array($step['event_type'], [ + HistoryEventType::MemoUpserted, + HistoryEventType::SearchAttributesUpserted, + HistoryEventType::SignalApplied, + ], true) + ) { + return true; + } + } + + return false; + } + /** * @param iterable<\Workflow\V2\Models\WorkflowMemo|\Workflow\V2\Models\WorkflowSearchAttribute> $rows * @param list, recorded_at: string|null}> $reusedSteps diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 1c48714da..78639eb69 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -12,6 +12,7 @@ use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Event; use Illuminate\Support\Facades\Queue; +use Illuminate\Support\Str; use Illuminate\Validation\ValidationException; use Mockery; use PHPUnit\Framework\Attributes\PreserveGlobalState; @@ -88,6 +89,7 @@ use Workflow\V2\Support\DefaultWorkflowTaskBridge; use Workflow\V2\Support\EmbeddedV2HistoryImport; use Workflow\V2\Support\HistoryExport; +use Workflow\V2\Support\MemoPayload; use Workflow\V2\Support\QueryStateReplayer; use Workflow\V2\Support\ReplayDiff; use Workflow\V2\Support\RunActivityView; @@ -256,6 +258,82 @@ public function testRedriveStartVisibilityCorpusCarriesToTheSuccessor(): void ); } + public function testRedriveServiceFenceCorpusRefusesUnqualifiedReuse(): void + { + $path = self::FIXTURE_DIR . '/redrive-service-unqualified-history.json'; + if (! is_file($path)) { + $this->markTestSkipped('The redrive service-fence evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['redrive_service_unqualified_history']; + self::stopWorkers(); + config([ + 'queue.default' => 'sync', + 'queue.connections.sync.driver' => 'sync', + ]); + $this->clearWorkflowState(); + $controlPlane = app(WorkflowControlPlane::class); + $options = [ + 'namespace' => 'regression-corpus-service', + 'queue' => 'service-workflows', + 'external_workflow_definition_fingerprint' => $scenario['worker_fingerprint'], + ]; + + $start = $controlPlane->start( + $scenario['service_workflow_type'], + 'regression-corpus-service-fence', + $options + [ + 'arguments' => Serializer::serializeWithCodec('avro', $fixture['workflow']['arguments']), + ] + ); + $this->assertTrue($start['started']); + $source = WorkflowRun::query()->findOrFail($start['workflow_run_id']); + // Step 1: the worker's memo upsert. Step 2: an activity that fails the run. + WorkflowHistoryEvent::record($source, HistoryEventType::MemoUpserted, [ + 'sequence' => 1, + 'entries' => MemoPayload::mapEnvelope($scenario['memo']), + 'merged' => MemoPayload::mapEnvelope($scenario['memo']), + ]); + $activityId = (string) Str::ulid(); + WorkflowHistoryEvent::record($source, HistoryEventType::ActivityScheduled, [ + 'activity_execution_id' => $activityId, + 'activity_type' => 'remote.second', + 'sequence' => 2, + ]); + WorkflowHistoryEvent::record($source, HistoryEventType::ActivityFailed, [ + 'activity_execution_id' => $activityId, + 'activity_type' => 'remote.second', + 'sequence' => 2, + 'message' => 'temporary failure', + ]); + $task = WorkflowTask::query()->where('workflow_run_id', $source->id)->firstOrFail(); + $task->forceFill([ + 'status' => TaskStatus::Leased, + 'lease_owner' => 'service-worker', + 'lease_expires_at' => now() + ->addMinute(), + ])->save(); + $source->forceFill([ + 'status' => RunStatus::Waiting, + ])->save(); + $this->assertTrue(app(DefaultWorkflowTaskBridge::class)->complete($task->id, [[ + 'type' => 'fail_workflow', + 'message' => 'temporary failure', + 'failed_step_sequence' => 2, + 'failed_activity_execution_id' => $activityId, + ]])['completed']); + $runs = WorkflowRun::query()->count(); + $events = WorkflowHistoryEvent::query()->count(); + + $result = $controlPlane->redrive('regression-corpus-service-fence', $source->id, $options); + + $this->assertSame($scenario['redrive'] === 'accepted', $result['accepted']); + $this->assertSame($scenario['reason'], $result['reason']); + $this->assertSame($runs, WorkflowRun::query()->count()); + $this->assertSame($events, WorkflowHistoryEvent::query()->count()); + $this->assertSame($source->id, $source->instance->fresh()->current_run_id); + } + public function testRedriveReusedMarkersCorpusKeepsTheirStepsAndTime(): void { $path = self::FIXTURE_DIR . '/redrive-reused-markers-clock.json'; diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-service-unqualified-history.json b/tests/Fixtures/V2/ReplayRegression/redrive-service-unqualified-history.json new file mode 100644 index 000000000..756a17954 --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-service-unqualified-history.json @@ -0,0 +1,39 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-service-unqualified-history", + "protocol_version": "1.0", + "bindings": ["php"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveWorkflow", + "arguments": ["Service Worker"], + "payload_codec": "avro" + }, + "redrive_service_unqualified_history": { + "service_workflow_type": "remote.redrive", + "worker_fingerprint": "worker-definition-1", + "reused_step": "MemoUpserted", + "memo": {"stage": "review"}, + "redrive": "refused", + "reason": "service_redrive_history_not_qualified" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": {} + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity", + "activity_type": "Tests\\Fixtures\\V2\\TestRedriveFirstActivity", + "payload_codec": "avro" + } + ] + } +} diff --git a/tests/Unit/V2/DefaultWorkflowControlPlaneRedriveRejectionTest.php b/tests/Unit/V2/DefaultWorkflowControlPlaneRedriveRejectionTest.php index ff7b2d05b..51fbd3d21 100644 --- a/tests/Unit/V2/DefaultWorkflowControlPlaneRedriveRejectionTest.php +++ b/tests/Unit/V2/DefaultWorkflowControlPlaneRedriveRejectionTest.php @@ -27,12 +27,15 @@ use Workflow\V2\Models\WorkflowLink; use Workflow\V2\Models\WorkflowMemo; use Workflow\V2\Models\WorkflowRun; +use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowSearchAttribute; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Models\WorkflowTimelineEntry; use Workflow\V2\Support\DefaultWorkflowTaskBridge; use Workflow\V2\Support\ExternalPayloadReference; use Workflow\V2\Support\ExternalPayloads; use Workflow\V2\Support\FailedRunRedrivePlan; +use Workflow\V2\Support\MemoPayload; use Workflow\V2\WorkflowStub; final class DefaultWorkflowControlPlaneRedriveRejectionTest extends TestCase @@ -210,6 +213,98 @@ public function testAWorkerSourcedRunRefusesADefinitionChangeAfterALocalClassApp $this->assertSame($before, $this->records()); } + /** + * @return iterable + */ + public static function serviceUnqualifiedSteps(): iterable + { + yield 'memo upsert' => ['memo']; + yield 'search attribute upsert' => ['search_attributes']; + yield 'signal wait' => ['signal']; + } + + #[DataProvider('serviceUnqualifiedSteps')] + public function testAServiceRunRefusesReusedStepsItsSdksAreNotQualifiedToReplay(string $step): void + { + $start = $this->controlPlane->start('remote.redrive', 'redrive-service-fence', [ + 'namespace' => 'tenant-redrive', + 'queue' => 'service-workflows', + 'arguments' => Serializer::serializeWithCodec('avro', ['Taylor']), + 'external_workflow_definition_fingerprint' => 'worker-definition-1', + ]); + $this->assertTrue($start['started']); + $source = WorkflowRun::query()->findOrFail($start['workflow_run_id']); + // Step 1 is the reused step; step 2 is the failed activity. + match ($step) { + 'memo' => WorkflowHistoryEvent::record($source, HistoryEventType::MemoUpserted, [ + 'sequence' => 1, + 'entries' => MemoPayload::mapEnvelope([ + 'stage' => 'review', + ]), + 'merged' => MemoPayload::mapEnvelope([ + 'stage' => 'review', + ]), + ]), + 'search_attributes' => WorkflowHistoryEvent::record($source, HistoryEventType::SearchAttributesUpserted, [ + 'sequence' => 1, + 'attributes' => [ + 'Stage' => 'review', + ], + 'attribute_types' => [ + 'Stage' => 'keyword', + ], + 'merged' => [ + 'Stage' => 'review', + ], + ]), + 'signal' => $this->recordAppliedSignal($source), + }; + $this->recordFailure($source, includeCompletedPrefix: false); + $source = $source->fresh(); + $before = $this->records(); + $projections = $this->projections(); + + $result = $this->controlPlane->redrive($source->workflow_instance_id, $source->id, $this->redriveOptions()); + + $this->assertFalse($result['accepted']); + $this->assertSame('service_redrive_history_not_qualified', $result['reason']); + $this->assertSame($before, $this->records()); + $this->assertSame($projections, $this->projections()); + $this->assertSame( + 'service_redrive_history_not_qualified', + $this->controlPlane->describe($source->workflow_instance_id, [ + 'run_id' => $source->id, + 'namespace' => 'tenant-redrive', + ])['actions']['redrive_blocked_reason'] ?? null, + ); + } + + public function testAServiceRunStillRedrivesAQualifiedReusedSideEffect(): void + { + $start = $this->controlPlane->start('remote.redrive', 'redrive-service-side-effect', [ + 'namespace' => 'tenant-redrive', + 'queue' => 'service-workflows', + 'arguments' => Serializer::serializeWithCodec('avro', ['Taylor']), + 'external_workflow_definition_fingerprint' => 'worker-definition-1', + ]); + $source = WorkflowRun::query()->findOrFail($start['workflow_run_id']); + WorkflowHistoryEvent::record($source, HistoryEventType::SideEffectRecorded, [ + 'sequence' => 1, + 'result' => Serializer::serializeWithCodec('avro', 'token'), + ]); + $this->recordFailure($source, includeCompletedPrefix: false); + + $result = $this->controlPlane->redrive($source->workflow_instance_id, $source->id, $this->redriveOptions()); + + $this->assertTrue($result['accepted'], (string) $result['reason']); + $this->assertSame(2, $result['resume_step_sequence']); + $reused = $this->event( + WorkflowRun::query()->findOrFail($result['workflow_run_id']), + HistoryEventType::SideEffectRecorded, + ); + $this->assertSame($source->id, $reused->payload['reused_from_run_id'] ?? null); + } + public function testServiceRunMemoAndSearchAttributeRowsCarryToTheRedriveSuccessor(): void { $source = $this->failedSource([ @@ -625,6 +720,52 @@ private function recordFailure(WorkflowRun $source, bool $includeCompletedPrefix $this->assertSame(RunStatus::Failed, $source->fresh()->status); } + private function recordAppliedSignal(WorkflowRun $source): WorkflowHistoryEvent + { + $waitId = (string) Str::ulid(); + $signalId = (string) Str::ulid(); + $commandId = (string) Str::ulid(); + WorkflowHistoryEvent::record($source, HistoryEventType::SignalWaitOpened, [ + 'signal_name' => 'approve', + 'signal_wait_id' => $waitId, + 'sequence' => 1, + ]); + WorkflowHistoryEvent::record($source, HistoryEventType::SignalReceived, [ + 'workflow_command_id' => $commandId, + 'signal_id' => $signalId, + 'workflow_instance_id' => $source->workflow_instance_id, + 'workflow_run_id' => $source->id, + 'signal_name' => 'approve', + 'signal_wait_id' => $waitId, + 'arguments' => Serializer::serializeWithCodec('avro', ['Grace']), + 'payload_codec' => 'avro', + ]); + + return WorkflowHistoryEvent::record($source, HistoryEventType::SignalApplied, [ + 'workflow_command_id' => $commandId, + 'signal_id' => $signalId, + 'signal_name' => 'approve', + 'signal_wait_id' => $waitId, + 'sequence' => 1, + 'value' => Serializer::serializeWithCodec('avro', 'Grace'), + ]); + } + + /** + * @return array>> + */ + private function projections(): array + { + $projections = []; + foreach ([WorkflowRunSummary::class, WorkflowTimelineEntry::class] as $model) { + $projections[$model] = $model::query()->orderBy('id')->get()->map( + static fn (Model $row): array => $row->getRawOriginal(), + )->all(); + } + + return $projections; + } + private function event(WorkflowRun $run, HistoryEventType $type): WorkflowHistoryEvent { return $run->historyEvents()