From 41ed11efbee5eff7225423e8856bcfd1863ae390 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 15:17:59 -0600 Subject: [PATCH 1/3] fix: report a redrive link as typed lineage (#864) Run detail and describe showed a redrive link as a diagnostic-only fallback on both sides, so Waterline labelled it a plain Parent or Child. The successor's WorkflowStarted already records recovery_kind redrive, continued_from_run_id and workflow_link_id. - RunLineageView::parentsForRun() adds a typed redrive parent from the successor's own WorkflowStarted. - RunLineageView::continuedWorkflowsForRun() adds a typed redrive successor for the failed run only when the successor's WorkflowStarted names that run and that link; otherwise the link stays a diagnostic fallback. Counterfactual: the typed entries fail on main; a link the successor does not confirm stays diagnostic. Replay corpus evidence: redrive-typed-lineage. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 8 ++ src/V2/Support/RunLineageView.php | 95 ++++++++++++++++++- .../V2EmbeddedReplayRegressionCorpusTest.php | 47 +++++++++ tests/Feature/V2/V2FailedRunRedriveTest.php | 61 ++++++++++++ .../redrive-typed-lineage.json | 38 ++++++++ 5 files changed, 248 insertions(+), 1 deletion(-) create mode 100644 tests/Fixtures/V2/ReplayRegression/redrive-typed-lineage.json diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d3e5a435..ed2e5805d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +- Run detail and describe report a redrive link as typed lineage. The + successor's `parents` and the failed run's `continuedWorkflows` show + `link_type: redrive` with `history_authority: typed_history` and + `diagnostic_only: false`, because the successor's `WorkflowStarted` records + `recovery_kind: redrive`, `continued_from_run_id` and `workflow_link_id`. + Before, both sides showed a diagnostic-only fallback. A link the successor's + history does not confirm stays a diagnostic fallback (#864). + ## 2.19.0 - 2026-11-03 - Rollback: releases before 2.19.0 ignore `reused_recorded_at` on the signal, diff --git a/src/V2/Support/RunLineageView.php b/src/V2/Support/RunLineageView.php index 2a8c37f19..902289307 100644 --- a/src/V2/Support/RunLineageView.php +++ b/src/V2/Support/RunLineageView.php @@ -24,7 +24,11 @@ public static function parentsForRun(WorkflowRun $run): array $entries = []; $seen = []; - foreach ([self::continuedParentEntry($run), self::childParentEntry($run)] as $entry) { + foreach ([ + self::continuedParentEntry($run), + self::redriveParentEntry($run), + self::childParentEntry($run), + ] as $entry) { if ($entry !== null) { $entries[] = $entry; $seen[self::key((string) $entry['link_type'], (string) $entry['workflow_run_id'])] = true; @@ -115,6 +119,11 @@ public static function continuedWorkflowsForRun(WorkflowRun $run): array } } + foreach (self::redriveSuccessorEntries($run) as $entry) { + $entries[] = $entry; + $seen[self::key('redrive', (string) $entry['workflow_run_id'])] = true; + } + foreach (self::childSequences($run) as $sequence) { $childRun = ChildRunHistory::childRunForSequence($run, $sequence); if ($childRun === null && ImportedLineageReferences::imported($run)) { @@ -277,6 +286,90 @@ private static function continuedParentEntry(WorkflowRun $run): ?array return null; } + /** + * A redrive successor names the failed run it continues in its own + * WorkflowStarted, so the link is typed history, not a fallback. + * + * @return array|null + */ + private static function redriveParentEntry(WorkflowRun $run): ?array + { + $started = self::workflowStartedEvent($run); + $payload = $started?->payload; + + if (! is_array($payload) || ($payload['recovery_kind'] ?? null) !== 'redrive') { + return null; + } + + $sourceRunId = self::stringValue($payload['continued_from_run_id'] ?? null); + $source = $sourceRunId === null ? null : self::resolveRun($run, $sourceRunId); + + if ($sourceRunId === null || (ImportedLineageReferences::imported($run) && $source === null)) { + return null; + } + + return self::entry( + id: self::stringValue($payload['workflow_link_id'] ?? null) ?? sprintf('redrive:%s', $sourceRunId), + linkType: 'redrive', + sequence: self::intValue($payload['resume_step_sequence'] ?? null), + isPrimaryParent: true, + relatedRun: $source, + instanceId: $run->workflow_instance_id, + runId: $sourceRunId, + createdAt: $started->recorded_at ?? $started->created_at, + historyAuthority: ChildRunHistory::HISTORY_AUTHORITY_TYPED, + ); + } + + /** + * The failed run's history is closed, so its redrive successor is typed + * only when the successor's WorkflowStarted names this run and this link. + * + * @return list> + */ + private static function redriveSuccessorEntries(WorkflowRun $run): array + { + $entries = []; + + foreach ($run->childLinks as $link) { + if ( + ! $link instanceof WorkflowLink + || $link->link_type !== 'redrive' + || $link->parent_workflow_run_id !== $run->id + || $link->child_workflow_instance_id !== $run->workflow_instance_id + ) { + continue; + } + + $successor = self::resolveRun($run, $link->child_workflow_run_id); + $payload = $successor === null ? null : self::workflowStartedEvent($successor)?->payload; + + if ( + ! is_array($payload) + || ($payload['recovery_kind'] ?? null) !== 'redrive' + || ($payload['continued_from_run_id'] ?? null) !== $run->id + || ($payload['workflow_link_id'] ?? null) !== $link->id + ) { + continue; + } + + $entries[] = self::entry( + id: $link->id, + linkType: 'redrive', + sequence: $link->sequence, + isPrimaryParent: true, + relatedRun: $successor, + instanceId: $run->workflow_instance_id, + runId: $successor->id, + relationPrefix: 'child', + createdAt: $link->created_at, + historyAuthority: ChildRunHistory::HISTORY_AUTHORITY_TYPED, + ); + } + + return $entries; + } + /** * @return array|null */ diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 78639eb69..77e4d16ec 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -94,6 +94,7 @@ use Workflow\V2\Support\ReplayDiff; use Workflow\V2\Support\RunActivityView; use Workflow\V2\Support\RunDetailView; +use Workflow\V2\Support\RunLineageView; use Workflow\V2\Support\RunTimelineProjector; use Workflow\V2\Support\ScheduleManager; use Workflow\V2\Support\ServiceCallPrincipal; @@ -594,6 +595,52 @@ public function testRedriveCorrectedDefinitionCorpusReplaysThePrefixWithNewCode( ); } + public function testRedriveTypedLineageCorpusLinksBothRunsFromHistory(): void + { + $path = self::FIXTURE_DIR . '/redrive-typed-lineage.json'; + if (! is_file($path)) { + $this->markTestSkipped('The redrive typed-lineage evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['redrive_typed_lineage']; + self::stopWorkers(); + config([ + 'queue.default' => 'sync', + 'queue.connections.sync.driver' => 'sync', + ]); + Cache::forget('test:redrive:first-calls'); + Cache::forget('test:redrive:second-calls'); + $this->clearWorkflowState(); + WorkflowStub::fake(); + + $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-redrive-lineage'); + $stub->start(...$fixture['workflow']['arguments']); + $this->assertTrue($stub->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($stub->runId()); + $result = app(WorkflowControlPlane::class)->redrive('regression-corpus-redrive-lineage', $source->id); + $this->assertTrue($result['accepted'], (string) $result['reason']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + + foreach ([ + [RunLineageView::parentsForRun($successor->fresh()), $source->id, $scenario['source_run_number']], + [ + RunLineageView::continuedWorkflowsForRun($source->fresh()), + $successor->id, + $scenario['successor_run_number'], + ], + ] as [$entries, $runId, $runNumber]) { + $redrive = array_values(array_filter( + $entries, + static fn (array $entry): bool => $entry['link_type'] === $scenario['link_type'], + )); + $this->assertCount(1, $redrive); + $this->assertSame($runId, $redrive[0]['workflow_run_id']); + $this->assertSame($runNumber, $redrive[0]['run_number']); + $this->assertSame($scenario['history_authority'], $redrive[0]['history_authority']); + $this->assertSame($scenario['diagnostic_only'], $redrive[0]['diagnostic_only']); + } + } + public function testRedriveReusedTimerCorpusDoesNotWaitAgain(): void { $path = self::FIXTURE_DIR . '/redrive-reused-timer-clock.json'; diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index 585fadd47..4eccb7b21 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -1146,6 +1146,67 @@ public function testALaterRedriveOnTheSameDefinitionRecordsNoInheritedChange(): ])['run']['definition_changed_from'] ?? null); } + public function testRunDetailReportsRedriveLineageAsTypedHistory(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveWorkflow::class, 'redrive-lineage-1'); + $workflow->start('Taylor'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + $result = app(WorkflowControlPlane::class)->redrive('redrive-lineage-1', $source->id); + $this->assertTrue($result['accepted'], (string) $result['reason']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $link = WorkflowLink::query()->where('link_type', 'redrive')->sole(); + + $parents = array_values(array_filter( + RunDetailView::forRun($successor->fresh())['parents'], + static fn (array $entry): bool => $entry['link_type'] === 'redrive', + )); + $this->assertCount(1, $parents); + $this->assertSame($link->id, $parents[0]['id']); + $this->assertSame($source->id, $parents[0]['workflow_run_id']); + $this->assertSame(1, $parents[0]['run_number']); + $this->assertFalse($parents[0]['diagnostic_only']); + $this->assertSame('typed_history', $parents[0]['history_authority']); + + $successors = array_values(array_filter( + RunDetailView::forRun($source->fresh())['continuedWorkflows'], + static fn (array $entry): bool => $entry['link_type'] === 'redrive', + )); + $this->assertCount(1, $successors); + $this->assertSame($successor->id, $successors[0]['workflow_run_id']); + $this->assertSame(2, $successors[0]['run_number']); + $this->assertFalse($successors[0]['diagnostic_only']); + $this->assertSame('typed_history', $successors[0]['history_authority']); + } + + public function testARedriveLinkTheSuccessorDoesNotConfirmStaysADiagnosticFallback(): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveWorkflow::class, 'redrive-lineage-2'); + $workflow->start('Taylor'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + $result = app(WorkflowControlPlane::class)->redrive('redrive-lineage-2', $source->id); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + // The successor's history names a different link. + $started = $this->sourceEvent($successor, HistoryEventType::WorkflowStarted); + $started->forceFill([ + 'payload' => array_merge($started->payload, [ + 'workflow_link_id' => 'another-link', + ]), + ])->save(); + + $successors = array_values(array_filter( + RunDetailView::forRun($source->fresh())['continuedWorkflows'], + static fn (array $entry): bool => $entry['link_type'] === 'redrive', + )); + $this->assertCount(1, $successors); + $this->assertTrue($successors[0]['diagnostic_only']); + } + public function testRedriveReusesOneMarkerForARepeatedPatchCall(): void { WorkflowStub::fake(); diff --git a/tests/Fixtures/V2/ReplayRegression/redrive-typed-lineage.json b/tests/Fixtures/V2/ReplayRegression/redrive-typed-lineage.json new file mode 100644 index 000000000..b08b89f3b --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/redrive-typed-lineage.json @@ -0,0 +1,38 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-typed-lineage", + "protocol_version": "1.0", + "bindings": ["php"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestRedriveWorkflow", + "arguments": ["Lineage"], + "payload_codec": "avro" + }, + "redrive_typed_lineage": { + "link_type": "redrive", + "history_authority": "typed_history", + "diagnostic_only": false, + "source_run_number": 1, + "successor_run_number": 2 + }, + "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" + } + ] + } +} From eb5dd38ac4338b4bcb53da626bc64cb001e8b0d0 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 15:51:41 -0600 Subject: [PATCH 2/3] Bind typed redrive lineage to the successor's exact start identity Security's hold on 41ed11ef. Both views now share one check of the successor's WorkflowStarted: it must name the successor's own run and instance, an earlier run of the same instance as the source, a nonblank link id and a positive resume step. The typed entry takes its step and time from that event, not the mutable link row. Anything less leaves the link row as a single diagnostic fallback. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 8 +- src/V2/Support/RunLineageView.php | 88 ++++++++++++--- tests/Feature/V2/V2FailedRunRedriveTest.php | 115 ++++++++++++++++++++ 3 files changed, 192 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ed2e5805d..92434175a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,8 +7,12 @@ `link_type: redrive` with `history_authority: typed_history` and `diagnostic_only: false`, because the successor's `WorkflowStarted` records `recovery_kind: redrive`, `continued_from_run_id` and `workflow_link_id`. - Before, both sides showed a diagnostic-only fallback. A link the successor's - history does not confirm stays a diagnostic fallback (#864). + Both sides take the step (`resume_step_sequence`) and time from that event, + not from the link row. The start must name the successor's own run and + instance, an earlier run of the same instance as the source, a nonblank link + id and a positive resume step. Before, both sides showed a diagnostic-only + fallback. A link the successor's history does not confirm stays a single + diagnostic fallback (#864). ## 2.19.0 - 2026-11-03 diff --git a/src/V2/Support/RunLineageView.php b/src/V2/Support/RunLineageView.php index 902289307..b0fffba86 100644 --- a/src/V2/Support/RunLineageView.php +++ b/src/V2/Support/RunLineageView.php @@ -294,29 +294,32 @@ private static function continuedParentEntry(WorkflowRun $run): ?array */ private static function redriveParentEntry(WorkflowRun $run): ?array { - $started = self::workflowStartedEvent($run); - $payload = $started?->payload; + $start = self::redriveStart($run); - if (! is_array($payload) || ($payload['recovery_kind'] ?? null) !== 'redrive') { + if ($start === null) { return null; } - $sourceRunId = self::stringValue($payload['continued_from_run_id'] ?? null); - $source = $sourceRunId === null ? null : self::resolveRun($run, $sourceRunId); + // The source is an earlier run of this instance. + $source = self::resolveRun($run, $start['source_run_id']); - if ($sourceRunId === null || (ImportedLineageReferences::imported($run) && $source === null)) { + if ( + $source === null + || $source->workflow_instance_id !== $run->workflow_instance_id + || (int) $source->run_number >= (int) $run->run_number + ) { return null; } return self::entry( - id: self::stringValue($payload['workflow_link_id'] ?? null) ?? sprintf('redrive:%s', $sourceRunId), + id: $start['link_id'], linkType: 'redrive', - sequence: self::intValue($payload['resume_step_sequence'] ?? null), + sequence: $start['resume_step_sequence'], isPrimaryParent: true, relatedRun: $source, instanceId: $run->workflow_instance_id, - runId: $sourceRunId, - createdAt: $started->recorded_at ?? $started->created_at, + runId: $start['source_run_id'], + createdAt: $start['created_at'], historyAuthority: ChildRunHistory::HISTORY_AUTHORITY_TYPED, ); } @@ -324,6 +327,7 @@ private static function redriveParentEntry(WorkflowRun $run): ?array /** * The failed run's history is closed, so its redrive successor is typed * only when the successor's WorkflowStarted names this run and this link. + * The entry takes its step and time from that event, not the link row. * * @return list> */ @@ -336,19 +340,22 @@ private static function redriveSuccessorEntries(WorkflowRun $run): array ! $link instanceof WorkflowLink || $link->link_type !== 'redrive' || $link->parent_workflow_run_id !== $run->id + || $link->parent_workflow_instance_id !== $run->workflow_instance_id || $link->child_workflow_instance_id !== $run->workflow_instance_id ) { continue; } $successor = self::resolveRun($run, $link->child_workflow_run_id); - $payload = $successor === null ? null : self::workflowStartedEvent($successor)?->payload; + $start = $successor === null ? null : self::redriveStart($successor); if ( - ! is_array($payload) - || ($payload['recovery_kind'] ?? null) !== 'redrive' - || ($payload['continued_from_run_id'] ?? null) !== $run->id - || ($payload['workflow_link_id'] ?? null) !== $link->id + $successor === null + || $start === null + || $successor->workflow_instance_id !== $run->workflow_instance_id + || (int) $successor->run_number <= (int) $run->run_number + || $start['source_run_id'] !== $run->id + || $start['link_id'] !== $link->id ) { continue; } @@ -356,13 +363,13 @@ private static function redriveSuccessorEntries(WorkflowRun $run): array $entries[] = self::entry( id: $link->id, linkType: 'redrive', - sequence: $link->sequence, + sequence: $start['resume_step_sequence'], isPrimaryParent: true, relatedRun: $successor, instanceId: $run->workflow_instance_id, runId: $successor->id, relationPrefix: 'child', - createdAt: $link->created_at, + createdAt: $start['created_at'], historyAuthority: ChildRunHistory::HISTORY_AUTHORITY_TYPED, ); } @@ -370,6 +377,53 @@ private static function redriveSuccessorEntries(WorkflowRun $run): array return $entries; } + /** + * The typed redrive identity a run's WorkflowStarted records: it names + * this exact run and instance, the failed source, the redrive link and the + * step the run resumes at. Anything less is not typed lineage. + * + * @return array{source_run_id: string, link_id: string, resume_step_sequence: int, created_at: CarbonInterface|null}|null + */ + private static function redriveStart(WorkflowRun $run): ?array + { + $started = self::workflowStartedEvent($run); + $payload = $started?->payload; + + if (! is_array($payload) || ($payload['recovery_kind'] ?? null) !== 'redrive') { + return null; + } + + $sourceRunId = self::nonBlank($payload['continued_from_run_id'] ?? null); + $linkId = self::nonBlank($payload['workflow_link_id'] ?? null); + $resume = $payload['resume_step_sequence'] ?? null; + + if ( + ($payload['workflow_run_id'] ?? null) !== $run->id + || ($payload['workflow_instance_id'] ?? null) !== $run->workflow_instance_id + || $sourceRunId === null + || $sourceRunId === $run->id + || $linkId === null + || ! is_int($resume) + || $resume < 1 + ) { + return null; + } + + return [ + 'source_run_id' => $sourceRunId, + 'link_id' => $linkId, + 'resume_step_sequence' => $resume, + 'created_at' => $started->recorded_at ?? $started->created_at, + ]; + } + + private static function nonBlank(mixed $value): ?string + { + return is_string($value) && trim($value) !== '' + ? $value + : null; + } + /** * @return array|null */ diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index 4eccb7b21..8c0307710 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -1179,6 +1179,121 @@ public function testRunDetailReportsRedriveLineageAsTypedHistory(): void $this->assertSame(2, $successors[0]['run_number']); $this->assertFalse($successors[0]['diagnostic_only']); $this->assertSame('typed_history', $successors[0]['history_authority']); + + // Both views take the step and time from the successor's WorkflowStarted, + // not from the mutable link row. + $started = $this->sourceEvent($successor, HistoryEventType::WorkflowStarted); + $link->forceFill([ + 'sequence' => 99, + 'created_at' => now() + ->addDay(), + ])->save(); + foreach ([ + RunDetailView::forRun($successor->fresh())['parents'], + RunDetailView::forRun($source->fresh())['continuedWorkflows'], + ] as $view) { + $redrive = array_values( + array_filter($view, static fn (array $entry): bool => $entry['link_type'] === 'redrive') + ); + $this->assertCount(1, $redrive); + $this->assertSame('typed_history', $redrive[0]['history_authority']); + $this->assertSame($result['resume_step_sequence'], $redrive[0]['sequence']); + $this->assertSame($started->payload['resume_step_sequence'], $redrive[0]['sequence']); + $this->assertSame( + ($started->recorded_at ?? $started->created_at)?->toJSON(), + $redrive[0]['created_at'], + ); + } + } + + /** + * @return iterable + */ + public static function redriveStartIdentityMutations(): iterable + { + // [mutation, whether the successor's own parent view may stay typed] + yield 'missing run id' => ['missing_run', false]; + yield 'another run id' => ['foreign_run', false]; + yield 'missing instance id' => ['missing_instance', false]; + yield 'another instance id' => ['foreign_instance', false]; + yield 'missing link id' => ['missing_link', false]; + yield 'blank link id' => ['blank_link', false]; + // The successor's history is the authority for its own link id; only + // the source side, which must match the link row, falls back. + yield 'another link id' => ['foreign_link', true]; + yield 'missing resume step' => ['missing_resume', false]; + yield 'zero resume step' => ['zero_resume', false]; + yield 'string resume step' => ['string_resume', false]; + yield 'missing source run' => ['missing_source', false]; + yield 'another source run' => ['foreign_source', false]; + yield 'source names the successor' => ['self_source', false]; + yield 'not a redrive' => ['not_redrive', false]; + } + + #[\PHPUnit\Framework\Attributes\DataProvider('redriveStartIdentityMutations')] + public function testARedriveStartWithoutItsExactIdentityIsNotTypedLineage(string $mutation, bool $parentTyped): void + { + WorkflowStub::fake(); + + $workflow = WorkflowStub::make(TestRedriveWorkflow::class, 'redrive-lineage-identity'); + $workflow->start('Taylor'); + $this->assertTrue($workflow->refresh()->failed()); + $source = WorkflowRun::query()->findOrFail($workflow->runId()); + $result = app(WorkflowControlPlane::class)->redrive('redrive-lineage-identity', $source->id); + $this->assertTrue($result['accepted'], (string) $result['reason']); + $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']); + $started = $this->sourceEvent($successor, HistoryEventType::WorkflowStarted); + $payload = $started->payload; + match ($mutation) { + 'missing_run' => (static function () use (&$payload): void { + unset($payload['workflow_run_id']); + })(), + 'foreign_run' => $payload['workflow_run_id'] = 'another-run', + 'missing_instance' => (static function () use (&$payload): void { + unset($payload['workflow_instance_id']); + })(), + 'foreign_instance' => $payload['workflow_instance_id'] = 'another-instance', + 'missing_link' => (static function () use (&$payload): void { + unset($payload['workflow_link_id']); + })(), + 'blank_link' => $payload['workflow_link_id'] = ' ', + 'foreign_link' => $payload['workflow_link_id'] = 'another-link', + 'missing_resume' => (static function () use (&$payload): void { + unset($payload['resume_step_sequence']); + })(), + 'zero_resume' => $payload['resume_step_sequence'] = 0, + 'string_resume' => $payload['resume_step_sequence'] = (string) $payload['resume_step_sequence'], + 'missing_source' => (static function () use (&$payload): void { + unset($payload['continued_from_run_id']); + })(), + 'foreign_source' => $payload['continued_from_run_id'] = 'another-run', + 'self_source' => $payload['continued_from_run_id'] = $successor->id, + 'not_redrive' => (static function () use (&$payload): void { + unset($payload['recovery_kind']); + })(), + }; + $started->forceFill([ + 'payload' => $payload, + ])->save(); + + foreach ([ + 'parent' => [RunDetailView::forRun($successor->fresh())['parents'], $parentTyped], + 'successor' => [RunDetailView::forRun($source->fresh())['continuedWorkflows'], false], + ] as $side => [$view, $typed]) { + $redrive = array_values( + array_filter($view, static fn (array $entry): bool => $entry['link_type'] === 'redrive') + ); + // One edge either way: the typed entry, or the link row as a + // diagnostic fallback. + $this->assertCount(1, $redrive, $side); + $this->assertSame( + $typed ? 'typed_history' : 'mutable_open_fallback', + $redrive[0]['history_authority'], + $side + ); + $this->assertSame(! $typed, $redrive[0]['diagnostic_only'], $side); + $this->assertSame($side === 'parent' ? $source->id : $successor->id, $redrive[0]['workflow_run_id'], $side); + } } public function testARedriveLinkTheSuccessorDoesNotConfirmStaysADiagnosticFallback(): void From 091b6da93502fe463f4766844fd7c4fca4fbe0b8 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 16:11:14 -0600 Subject: [PATCH 3/3] Cross-confirm typed redrive parent against its exact link row Security's hold on eb5dd38a. The successor's typed parent now needs exactly one redrive link whose id is the start's workflow_link_id, whose parent run and instance are the named source, and whose child run and instance are this successor. A foreign link id, or a start naming another real earlier failed run of the instance, leaves the one existing link row as a single diagnostic edge on both views. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 3 +- src/V2/Support/RunLineageView.php | 18 ++- tests/Feature/V2/V2FailedRunRedriveTest.php | 129 ++++++++++++++------ 3 files changed, 112 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 92434175a..23a6ce8ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,8 @@ Both sides take the step (`resume_step_sequence`) and time from that event, not from the link row. The start must name the successor's own run and instance, an earlier run of the same instance as the source, a nonblank link - id and a positive resume step. Before, both sides showed a diagnostic-only + id and a positive resume step, and exactly one redrive link row must join + that source to the successor under that link id. Before, both sides showed a diagnostic-only fallback. A link the successor's history does not confirm stays a single diagnostic fallback (#864). diff --git a/src/V2/Support/RunLineageView.php b/src/V2/Support/RunLineageView.php index b0fffba86..337617641 100644 --- a/src/V2/Support/RunLineageView.php +++ b/src/V2/Support/RunLineageView.php @@ -300,7 +300,8 @@ private static function redriveParentEntry(WorkflowRun $run): ?array return null; } - // The source is an earlier run of this instance. + // The source is an earlier run of this instance, and exactly one + // redrive link joins it to this run under the recorded link id. $source = self::resolveRun($run, $start['source_run_id']); if ( @@ -311,6 +312,20 @@ private static function redriveParentEntry(WorkflowRun $run): ?array return null; } + $links = $run->parentLinks->filter( + static fn (mixed $link): bool => $link instanceof WorkflowLink + && $link->link_type === 'redrive' + && $link->id === $start['link_id'] + && $link->parent_workflow_run_id === $source->id + && $link->parent_workflow_instance_id === $run->workflow_instance_id + && $link->child_workflow_run_id === $run->id + && $link->child_workflow_instance_id === $run->workflow_instance_id + ); + + if ($links->count() !== 1) { + return null; + } + return self::entry( id: $start['link_id'], linkType: 'redrive', @@ -352,6 +367,7 @@ private static function redriveSuccessorEntries(WorkflowRun $run): array if ( $successor === null || $start === null + || $successor->id !== $link->child_workflow_run_id || $successor->workflow_instance_id !== $run->workflow_instance_id || (int) $successor->run_number <= (int) $run->run_number || $start['source_run_id'] !== $run->id diff --git a/tests/Feature/V2/V2FailedRunRedriveTest.php b/tests/Feature/V2/V2FailedRunRedriveTest.php index 8c0307710..1437d58d4 100644 --- a/tests/Feature/V2/V2FailedRunRedriveTest.php +++ b/tests/Feature/V2/V2FailedRunRedriveTest.php @@ -1207,31 +1207,28 @@ public function testRunDetailReportsRedriveLineageAsTypedHistory(): void } /** - * @return iterable + * @return iterable */ public static function redriveStartIdentityMutations(): iterable { - // [mutation, whether the successor's own parent view may stay typed] - yield 'missing run id' => ['missing_run', false]; - yield 'another run id' => ['foreign_run', false]; - yield 'missing instance id' => ['missing_instance', false]; - yield 'another instance id' => ['foreign_instance', false]; - yield 'missing link id' => ['missing_link', false]; - yield 'blank link id' => ['blank_link', false]; - // The successor's history is the authority for its own link id; only - // the source side, which must match the link row, falls back. - yield 'another link id' => ['foreign_link', true]; - yield 'missing resume step' => ['missing_resume', false]; - yield 'zero resume step' => ['zero_resume', false]; - yield 'string resume step' => ['string_resume', false]; - yield 'missing source run' => ['missing_source', false]; - yield 'another source run' => ['foreign_source', false]; - yield 'source names the successor' => ['self_source', false]; - yield 'not a redrive' => ['not_redrive', false]; + yield 'missing run id' => ['missing_run']; + yield 'another run id' => ['foreign_run']; + yield 'missing instance id' => ['missing_instance']; + yield 'another instance id' => ['foreign_instance']; + yield 'missing link id' => ['missing_link']; + yield 'blank link id' => ['blank_link']; + yield 'another link id' => ['foreign_link']; + yield 'missing resume step' => ['missing_resume']; + yield 'zero resume step' => ['zero_resume']; + yield 'string resume step' => ['string_resume']; + yield 'missing source run' => ['missing_source']; + yield 'another source run' => ['foreign_source']; + yield 'source names the successor' => ['self_source']; + yield 'not a redrive' => ['not_redrive']; } #[\PHPUnit\Framework\Attributes\DataProvider('redriveStartIdentityMutations')] - public function testARedriveStartWithoutItsExactIdentityIsNotTypedLineage(string $mutation, bool $parentTyped): void + public function testARedriveStartWithoutItsExactIdentityIsNotTypedLineage(string $mutation): void { WorkflowStub::fake(); @@ -1276,24 +1273,56 @@ public function testARedriveStartWithoutItsExactIdentityIsNotTypedLineage(string 'payload' => $payload, ])->save(); - foreach ([ - 'parent' => [RunDetailView::forRun($successor->fresh())['parents'], $parentTyped], - 'successor' => [RunDetailView::forRun($source->fresh())['continuedWorkflows'], false], - ] as $side => [$view, $typed]) { - $redrive = array_values( - array_filter($view, static fn (array $entry): bool => $entry['link_type'] === 'redrive') - ); - // One edge either way: the typed entry, or the link row as a - // diagnostic fallback. - $this->assertCount(1, $redrive, $side); - $this->assertSame( - $typed ? 'typed_history' : 'mutable_open_fallback', - $redrive[0]['history_authority'], - $side - ); - $this->assertSame(! $typed, $redrive[0]['diagnostic_only'], $side); - $this->assertSame($side === 'parent' ? $source->id : $successor->id, $redrive[0]['workflow_run_id'], $side); + $this->assertOneDiagnosticRedriveEdge($source, $successor); + } + + /** + * A redrive chain A -> B -> C where C's start names A, a real earlier + * failed run of the same instance, instead of B. No link joins A and C, + * so neither run gets a typed or a second edge. + */ + public function testARedriveStartNamingAnotherEarlierFailedRunIsNotTypedLineage(): void + { + WorkflowStub::fake(); + config([ + 'workflows.v2.testing.redrive_failures_before_success' => 2, + ]); + $controlPlane = app(WorkflowControlPlane::class); + + $workflow = WorkflowStub::make(TestRedriveWorkflow::class, 'redrive-lineage-chain'); + $workflow->start('Taylor'); + $this->assertTrue($workflow->refresh()->failed()); + $first = WorkflowRun::query()->findOrFail($workflow->runId()); + $second = $controlPlane->redrive('redrive-lineage-chain', $first->id); + $this->assertTrue($second['accepted'], (string) $second['reason']); + $secondRun = WorkflowRun::query()->findOrFail($second['workflow_run_id']); + $this->assertSame(RunStatus::Failed, $secondRun->status); + $third = $controlPlane->redrive('redrive-lineage-chain', $secondRun->id); + $this->assertTrue($third['accepted'], (string) $third['reason']); + $thirdRun = WorkflowRun::query()->findOrFail($third['workflow_run_id']); + + // The unmodified chain is typed on every edge. + foreach ([[$first, $secondRun], [$secondRun, $thirdRun]] as [$source, $successor]) { + $parents = $this->redriveEdges(RunDetailView::forRun($successor->fresh())['parents']); + $this->assertCount(1, $parents); + $this->assertSame('typed_history', $parents[0]['history_authority']); + $this->assertSame($source->id, $parents[0]['workflow_run_id']); } + + $started = $this->sourceEvent($thirdRun, HistoryEventType::WorkflowStarted); + $started->forceFill([ + 'payload' => array_merge($started->payload, [ + 'continued_from_run_id' => $first->id, + ]), + ])->save(); + + $this->assertOneDiagnosticRedriveEdge($secondRun, $thirdRun); + // The first run keeps its typed edge to the second run and gains no + // edge to the third. + $firstSuccessors = $this->redriveEdges(RunDetailView::forRun($first->fresh())['continuedWorkflows']); + $this->assertCount(1, $firstSuccessors); + $this->assertSame($secondRun->id, $firstSuccessors[0]['workflow_run_id']); + $this->assertSame('typed_history', $firstSuccessors[0]['history_authority']); } public function testARedriveLinkTheSuccessorDoesNotConfirmStaysADiagnosticFallback(): void @@ -1666,6 +1695,34 @@ public function testFailedSourceAndRedriveSuccessorSurviveExportImport(): void ], json_decode($coldReplay->getOutput(), true, flags: JSON_THROW_ON_ERROR)); } + /** + * Both views show the one existing link row as a diagnostic fallback, and + * no typed redrive edge. + */ + private function assertOneDiagnosticRedriveEdge(WorkflowRun $source, WorkflowRun $successor): void + { + foreach ([ + 'parent' => [RunDetailView::forRun($successor->fresh())['parents'], $source->id], + 'successor' => [RunDetailView::forRun($source->fresh())['continuedWorkflows'], $successor->id], + ] as $side => [$view, $runId]) { + $redrive = $this->redriveEdges($view); + $this->assertCount(1, $redrive, $side); + $this->assertSame('mutable_open_fallback', $redrive[0]['history_authority'], $side); + $this->assertTrue($redrive[0]['diagnostic_only'], $side); + $this->assertSame($runId, $redrive[0]['workflow_run_id'], $side); + } + } + + /** + * @param list> $view + * + * @return list> + */ + private function redriveEdges(array $view): array + { + return array_values(array_filter($view, static fn (array $entry): bool => $entry['link_type'] === 'redrive')); + } + private function failedSignalSource(string $instanceId): WorkflowRun { $workflow = WorkflowStub::make(TestRedriveSignalWorkflow::class, $instanceId);