From 9ffb67901a98ad4048ab819bd47a5016ababc809 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 19:14:30 +0000 Subject: [PATCH 1/6] Reject cross-namespace history-import target conflicts --- src/V2/Support/EmbeddedV2HistoryImport.php | 27 +++++- tests/Unit/V2/EmbeddedV2HistoryImportTest.php | 93 +++++++++++++++++++ .../V2/EmbeddedV2ImportEligibilityTest.php | 4 +- 3 files changed, 120 insertions(+), 4 deletions(-) diff --git a/src/V2/Support/EmbeddedV2HistoryImport.php b/src/V2/Support/EmbeddedV2HistoryImport.php index b5819988c..b1bbfbe66 100644 --- a/src/V2/Support/EmbeddedV2HistoryImport.php +++ b/src/V2/Support/EmbeddedV2HistoryImport.php @@ -146,7 +146,12 @@ public static function import(array $bundle, array $options = []): array ]; } - $existing = self::existingRunOutcome($workflow['run_id'], self::stringValue($bundle['dedupe_key'] ?? null)); + $namespace = self::stringValue($workflow['namespace'] ?? null); + $existing = self::existingRunOutcome( + $workflow['run_id'], + self::stringValue($bundle['dedupe_key'] ?? null), + $namespace + ); if ($existing['status'] === 'already_imported') { return [ @@ -169,6 +174,20 @@ public static function import(array $bundle, array $options = []): array ]; } + $instance = WorkflowInstance::query()->find($workflow['instance_id']); + if ($instance instanceof WorkflowInstance && $instance->namespace !== $namespace) { + return [ + ...$base, + 'status' => 'rejected', + 'eligibility' => self::withEligibilityError( + $eligibility, + 'target.instance_namespace_conflict', + 'A workflow instance with this instance_id already exists outside the target namespace.', + ), + 'rows' => self::emptyRows(), + ]; + } + if ($dryRun) { return [ ...$base, @@ -240,6 +259,7 @@ private static function writeBundle( /** @var WorkflowInstance|null $instance */ $instance = WorkflowInstance::query() + ->where('namespace', $namespace) ->lockForUpdate() ->find($instanceId); @@ -1184,7 +1204,7 @@ private static function rejectOpenLeases(array $bundle, array &$errors): void /** * @return array{status: string} */ - private static function existingRunOutcome(?string $runId, ?string $dedupeKey): array + private static function existingRunOutcome(?string $runId, ?string $dedupeKey, ?string $namespace): array { if ($runId === null) { return [ @@ -1202,7 +1222,8 @@ private static function existingRunOutcome(?string $runId, ?string $dedupeKey): } if ( - $run->import_source === EmbeddedV2ImportContract::IMPORT_SOURCE + $run->namespace === $namespace + && $run->import_source === EmbeddedV2ImportContract::IMPORT_SOURCE && is_string($run->import_dedupe_key) && $run->import_dedupe_key === $dedupeKey ) { diff --git a/tests/Unit/V2/EmbeddedV2HistoryImportTest.php b/tests/Unit/V2/EmbeddedV2HistoryImportTest.php index 36514ea70..676539992 100644 --- a/tests/Unit/V2/EmbeddedV2HistoryImportTest.php +++ b/tests/Unit/V2/EmbeddedV2HistoryImportTest.php @@ -7,6 +7,7 @@ use Illuminate\Support\Carbon; use Illuminate\Support\Facades\DB; use Illuminate\Support\Str; +use PHPUnit\Framework\Attributes\DataProvider; use Tests\TestCase; use Workflow\Serializers\CodecRegistry; use Workflow\Serializers\Serializer; @@ -40,6 +41,81 @@ final class EmbeddedV2HistoryImportTest extends TestCase { + #[DataProvider('namespaceConflicts')] + public function testItNeverTreatsAnotherNamespacesRunAsAnIdempotentImport( + ?string $target, + ?string $existing, + bool $dryRun + ): void { + $bundle = $this->runningBundleWithOpenWork(); + $bundle['workflow']['namespace'] = $target; + $bundle = $this->resealBundle($bundle); + $this->clearWorkflowState(); + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundle)['status']); + WorkflowRun::query()->update([ + 'namespace' => $existing, + ]); + WorkflowInstance::query()->update([ + 'namespace' => $existing, + ]); + $before = $this->workflowStateSnapshot(); + + $report = EmbeddedV2HistoryImport::import($bundle, [ + 'dry_run' => $dryRun, + ]); + + $this->assertSame('rejected', $report['status']); + $this->assertContains('target.run_id_conflict', array_column($report['eligibility']['errors'], 'rule')); + $this->assertSame(0, array_sum($report['rows'])); + $this->assertSame($before, $this->workflowStateSnapshot()); + } + + #[DataProvider('namespaceConflicts')] + public function testItNeverAttachesAnImportedRunToAnotherNamespacesInstance( + ?string $target, + ?string $existing, + bool $dryRun + ): void { + $bundle = $this->runningBundleWithOpenWork(); + $bundle['workflow']['namespace'] = $target; + $bundle = $this->resealBundle($bundle); + $this->clearWorkflowState(); + WorkflowInstance::query()->create([ + 'id' => $bundle['workflow']['instance_id'], + 'namespace' => $existing, + 'workflow_class' => $bundle['workflow']['workflow_class'], + 'workflow_type' => $bundle['workflow']['workflow_type'], + 'run_count' => 7, + 'current_run_id' => (string) Str::ulid(), + ]); + $before = $this->workflowStateSnapshot(); + + $report = EmbeddedV2HistoryImport::import($bundle, [ + 'dry_run' => $dryRun, + ]); + + $this->assertSame('rejected', $report['status']); + $this->assertContains( + 'target.instance_namespace_conflict', + array_column($report['eligibility']['errors'], 'rule') + ); + $this->assertSame(0, array_sum($report['rows'])); + $this->assertSame($before, $this->workflowStateSnapshot()); + } + + public static function namespaceConflicts(): iterable + { + foreach ([ + 'named' => ['tenant-a', 'tenant-b'], + 'null-target' => [null, 'tenant-b'], + 'null-existing' => ['tenant-a', null], + ] as $label => [$target, $existing]) { + foreach ([false, true] as $dryRun) { + yield $label . ($dryRun ? '-preview' : '-write') => [$target, $existing, $dryRun]; + } + } + } + public function testContractPublishesImportRulesAndAuditMarkers(): void { $manifest = EmbeddedV2ImportContract::manifest(); @@ -601,6 +677,23 @@ public function testItImportsExternalizedRunCommandSignalAndUpdatePayloadEnvelop $this->assertSame([], $fallbackUpdate['validation_errors']); } + /** + * @return array + */ + private function workflowStateSnapshot(): array + { + $snapshot = []; + foreach (['workflow_instances', 'workflow_runs', 'workflow_tasks', 'workflow_history_events', + 'workflow_commands', 'workflow_signal_records', 'workflow_updates', 'workflow_links', + 'activity_executions', 'activity_attempts', 'workflow_run_timers', 'workflow_failures', + 'workflow_run_summaries', 'workflow_run_waits', 'workflow_run_timeline_entries', + 'workflow_run_lineage_entries', 'workflow_run_timer_entries', 'workflow_memos', 'workflow_search_attributes'] as $table) { + $snapshot[$table] = DB::table($table)->get()->map(static fn (object $row): array => (array) $row)->all(); + } + + return $snapshot; + } + /** * @return array */ diff --git a/tests/Unit/V2/EmbeddedV2ImportEligibilityTest.php b/tests/Unit/V2/EmbeddedV2ImportEligibilityTest.php index 2ecb7f9ed..bc7d772e7 100644 --- a/tests/Unit/V2/EmbeddedV2ImportEligibilityTest.php +++ b/tests/Unit/V2/EmbeddedV2ImportEligibilityTest.php @@ -168,7 +168,9 @@ public function testRunNumberConflictPreservesTheExistingInstanceAndRun(): void $original = $bundle; for ($delivery = 0; $delivery < 2; ++$delivery) { - $report = EmbeddedV2HistoryImport::import($bundle); + $report = EmbeddedV2HistoryImport::import($bundle, [ + 'namespace' => 'import-tests', + ]); $this->assertSame('rejected', $report['status']); $this->assertContains('target.write_failed', array_column($report['eligibility']['errors'], 'rule')); $this->assertSame(0, array_sum($report['rows'])); From 5c766e3188fb1fb7ac9ebb267773bcc5e37912f6 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 19:18:21 +0000 Subject: [PATCH 2/6] Cover import namespace changes after preflight --- tests/Unit/V2/EmbeddedV2HistoryImportTest.php | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/tests/Unit/V2/EmbeddedV2HistoryImportTest.php b/tests/Unit/V2/EmbeddedV2HistoryImportTest.php index 676539992..5ee6ba51f 100644 --- a/tests/Unit/V2/EmbeddedV2HistoryImportTest.php +++ b/tests/Unit/V2/EmbeddedV2HistoryImportTest.php @@ -116,6 +116,69 @@ public static function namespaceConflicts(): iterable } } + public function testTheTransactionalInstanceLookupRechecksNamespaceAfterPreflight(): void + { + $bundle = $this->runningBundleWithOpenWork(); + $bundle['workflow']['namespace'] = 'tenant-a'; + $bundle = $this->resealBundle($bundle); + $this->clearWorkflowState(); + WorkflowInstance::query()->create([ + 'id' => $bundle['workflow']['instance_id'], + 'namespace' => 'tenant-a', + 'workflow_class' => $bundle['workflow']['workflow_class'], + 'workflow_type' => $bundle['workflow']['workflow_type'], + 'run_count' => 7, + ]); + $afterTransfer = null; + $transferred = false; + DB::listen(function ($query) use (&$transferred, &$afterTransfer, $bundle): void { + if ($transferred || DB::transactionLevel() !== 0 + || ! str_contains($query->sql, 'select') || ! str_contains($query->sql, 'workflow_instances')) { + return; + } + // The result of the preflight read is already materialized. A + // concurrent transfer must not authorize attaching to its old scope. + $transferred = true; + WorkflowInstance::query()->whereKey($bundle['workflow']['instance_id'])->update([ + 'namespace' => 'tenant-b', + ]); + $afterTransfer = $this->workflowStateSnapshot(); + }); + + $report = EmbeddedV2HistoryImport::import($bundle); + + $this->assertTrue($transferred); + $this->assertSame('rejected', $report['status']); + $this->assertContains('target.write_failed', array_column($report['eligibility']['errors'], 'rule')); + $this->assertSame(0, array_sum($report['rows'])); + $this->assertSame($afterTransfer, $this->workflowStateSnapshot()); + } + + public function testItCanImportANewRunIntoAnExistingInstanceInTheSameNamespace(): void + { + $bundle = $this->runningBundleWithOpenWork(); + $bundle['workflow']['namespace'] = 'tenant-a'; + $bundle = $this->resealBundle($bundle); + $this->clearWorkflowState(); + WorkflowInstance::query()->create([ + 'id' => $bundle['workflow']['instance_id'], + 'namespace' => 'tenant-a', + 'workflow_class' => $bundle['workflow']['workflow_class'], + 'workflow_type' => $bundle['workflow']['workflow_type'], + 'run_count' => 7, + ]); + + $report = EmbeddedV2HistoryImport::import($bundle); + + $this->assertSame('imported', $report['status']); + $this->assertSame(0, $report['rows']['workflow_instances']); + $this->assertSame(1, $report['rows']['workflow_runs']); + $instance = WorkflowInstance::query()->findOrFail($bundle['workflow']['instance_id']); + $this->assertSame('tenant-a', $instance->namespace); + $this->assertSame(7, (int) $instance->run_count); + $this->assertSame($bundle['workflow']['run_id'], $instance->current_run_id); + } + public function testContractPublishesImportRulesAndAuditMarkers(): void { $manifest = EmbeddedV2ImportContract::manifest(); From f0e6e9be13de3d5d3dcdf5de42af720bc35ad8bd Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 19:41:44 +0000 Subject: [PATCH 3/6] Require imported messages to bind to their bundle commands --- src/V2/Support/EmbeddedV2HistoryImport.php | 2 + .../EmbeddedV2ImportMessageReferences.php | 79 +++++++ tests/Unit/V2/EmbeddedV2HistoryImportTest.php | 198 ++++++++++++++++++ 3 files changed, 279 insertions(+) create mode 100644 src/V2/Support/EmbeddedV2ImportMessageReferences.php diff --git a/src/V2/Support/EmbeddedV2HistoryImport.php b/src/V2/Support/EmbeddedV2HistoryImport.php index b1bbfbe66..c5f2537b7 100644 --- a/src/V2/Support/EmbeddedV2HistoryImport.php +++ b/src/V2/Support/EmbeddedV2HistoryImport.php @@ -998,6 +998,8 @@ private static function eligibility(array $bundle, array $integrity, array $work self::rejectOpenLeases($bundle, $errors); } + $errors = [...$errors, ...EmbeddedV2ImportMessageReferences::errors($bundle)]; + return [ 'eligible' => $errors === [], 'errors' => $errors, diff --git a/src/V2/Support/EmbeddedV2ImportMessageReferences.php b/src/V2/Support/EmbeddedV2ImportMessageReferences.php new file mode 100644 index 000000000..1f9d73d95 --- /dev/null +++ b/src/V2/Support/EmbeddedV2ImportMessageReferences.php @@ -0,0 +1,79 @@ + $bundle + * @return list + */ + public static function errors(array $bundle): array + { + $commands = []; + foreach (self::rows($bundle['commands'] ?? null) as $command) { + $id = self::string($command['id'] ?? null); + $type = CommandType::tryFrom(self::string($command['type'] ?? null) ?? ''); + $status = CommandStatus::tryFrom(self::string($command['status'] ?? null) ?? ''); + if ($id !== null && $type !== null && $status !== null) { + $commands[$id][] = $type->value; + } + } + + $errors = []; + foreach ([ + 'signals' => 'signal', + 'updates' => 'update', + ] as $section => $kind) { + foreach (self::rows($bundle[$section] ?? null) as $index => $message) { + $status = self::string($message['status'] ?? null) ?? ''; + $knownStatus = $kind === 'signal' ? SignalStatus::tryFrom($status) : UpdateStatus::tryFrom($status); + if ($knownStatus === null) { + continue; + } + $commandId = self::string($message['command_id'] ?? null); + if ($commandId !== null && ($commands[$commandId] ?? []) === [$kind]) { + continue; + } + // Commands are inserted first, for this exact run/instance, + // in the import transaction. A colliding existing ID fails + // that insert and rolls back before a message can bind to it. + $errors[] = [ + 'rule' => 'references.message_command_invalid', + 'message' => sprintf( + '%s.%d.command_id must identify exactly one %s command imported from this bundle.', + $section, + $index, + $kind, + ), + ]; + } + } + + return $errors; + } + + /** + * @return list> + */ + private static function rows(mixed $value): array + { + return is_array($value) ? array_values(array_filter($value, is_array(...))) : []; + } + + private static function string(mixed $value): ?string + { + if (! is_string($value)) { + return null; + } + return $value === '' ? null : $value; + } +} diff --git a/tests/Unit/V2/EmbeddedV2HistoryImportTest.php b/tests/Unit/V2/EmbeddedV2HistoryImportTest.php index 5ee6ba51f..cbb814a47 100644 --- a/tests/Unit/V2/EmbeddedV2HistoryImportTest.php +++ b/tests/Unit/V2/EmbeddedV2HistoryImportTest.php @@ -29,6 +29,7 @@ use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Models\WorkflowTimer; use Workflow\V2\Models\WorkflowUpdate; +use Workflow\V2\Support\DefaultWorkflowTaskBridge; use Workflow\V2\Support\EmbeddedV2HistoryImport; use Workflow\V2\Support\EmbeddedV2ImportContract; use Workflow\V2\Support\ExternalPayloadReference; @@ -41,6 +42,100 @@ final class EmbeddedV2HistoryImportTest extends TestCase { + #[DataProvider('invalidMessageCommands')] + public function testImportRejectsMessageCommandReferencesOutsideItsBundle( + string $kind, + string $case, + bool $dryRun + ): void { + [$bundle, $messageId] = $this->messageBundle($kind); + $this->clearWorkflowState(); + $command = null; + $foreignBefore = null; + if (in_array($case, ['foreign-namespace', 'same-namespace-other-run', 'null-namespace'], true)) { + $namespace = match ($case) { + 'same-namespace-other-run' => 'tenant-a', + 'null-namespace' => null, + default => 'tenant-b', + }; + $command = $this->otherMessageCommand($kind, $namespace); + $foreignBefore = $command->getAttributes(); + $bundle[$kind === 'signal' ? 'signals' : 'updates'][0]['command_id'] = $command->id; + } elseif ($case === 'missing-command') { + $bundle['commands'] = []; + } elseif ($case === 'wrong-kind') { + $bundle['commands'][0]['type'] = $kind === 'signal' ? 'update' : 'signal'; + } elseif ($case === 'invalid-command-status') { + $bundle['commands'][0]['status'] = 'unknown'; + } elseif ($case === 'padded-command-kind') { + $bundle['commands'][0]['type'] = ' ' . $kind . ' '; + } elseif ($case === 'padded-reference') { + $bundle[$kind === 'signal' ? 'signals' : 'updates'][0]['command_id'] = ' ' . $bundle['commands'][0]['id'] . ' '; + } else { + unset($bundle[$kind === 'signal' ? 'signals' : 'updates'][0]['command_id']); + } + $bundle = $this->resealBundle($bundle); + $before = $this->workflowStateSnapshot(); + + $report = EmbeddedV2HistoryImport::import($bundle, [ + 'dry_run' => $dryRun, + ]); + if (! $dryRun && $report['status'] === 'imported' && $command instanceof WorkflowCommand) { + // The counterfactual must reach ordinary worker application, where + // an unowned command ID used to change another run's command. + $this->applyImportedMessage($bundle, $kind, $messageId); + } + + if ($command instanceof WorkflowCommand) { + $this->assertSame($foreignBefore, $command->fresh()->getAttributes()); + } + $this->assertSame('rejected', $report['status']); + $this->assertContains( + 'references.message_command_invalid', + array_column($report['eligibility']['errors'], 'rule') + ); + $this->assertSame(0, array_sum($report['rows'])); + $this->assertSame($before, $this->workflowStateSnapshot()); + } + + public static function invalidMessageCommands(): iterable + { + foreach (['signal', 'update'] as $kind) { + foreach (['foreign-namespace', 'same-namespace-other-run', 'null-namespace', 'missing-command', + 'wrong-kind', 'invalid-command-status', 'missing-command-id', 'padded-command-kind', 'padded-reference'] as $case) { + foreach ([false, true] as $dryRun) { + yield $kind . '/' . $case . ($dryRun ? '/preview' : '/write') => [$kind, $case, $dryRun]; + } + } + } + } + + #[DataProvider('messageKinds')] + public function testOwnedImportedMessageCommandsStillApplyOnTheImportedRun(string $kind): void + { + [$bundle, $messageId] = $this->messageBundle($kind); + $this->clearWorkflowState(); + $other = $this->otherMessageCommand($kind, 'tenant-b'); + $before = $other->getAttributes(); + + $report = EmbeddedV2HistoryImport::import($bundle); + + $this->assertSame('imported', $report['status']); + $this->applyImportedMessage($bundle, $kind, $messageId); + $command = WorkflowCommand::query()->findOrFail($bundle['commands'][0]['id']); + $this->assertSame($bundle['workflow']['run_id'], $command->workflow_run_id); + $this->assertSame($bundle['workflow']['instance_id'], $command->workflow_instance_id); + $this->assertNotNull($command->applied_at); + $this->assertSame($before, $other->fresh()->getAttributes()); + $this->assertSame('already_imported', EmbeddedV2HistoryImport::import($bundle)['status']); + } + + public static function messageKinds(): iterable + { + yield 'signal' => ['signal']; + yield 'update' => ['update']; + } + #[DataProvider('namespaceConflicts')] public function testItNeverTreatsAnotherNamespacesRunAsAnIdempotentImport( ?string $target, @@ -740,6 +835,109 @@ public function testItImportsExternalizedRunCommandSignalAndUpdatePayloadEnvelop $this->assertSame([], $fallbackUpdate['validation_errors']); } + /** + * @return array{array, string} + */ + private function messageBundle(string $kind): array + { + $bundle = $this->runningBundleWithOpenWork(); + $bundle['workflow']['namespace'] = 'tenant-a'; + $commandId = (string) Str::ulid(); + $messageId = (string) Str::ulid(); + $bundle['commands'] = [[ + 'id' => $commandId, + 'type' => $kind, + 'status' => 'accepted', + 'sequence' => 1, + 'payload_codec' => CodecRegistry::defaultCodec(), + 'accepted_at' => now() + ->toJSON(), + ]]; + $bundle[$kind === 'signal' ? 'signals' : 'updates'] = [[ + 'id' => $messageId, + 'command_id' => $commandId, + 'name' => 'imported-message', + 'status' => $kind === 'signal' ? 'received' : 'accepted', + 'arguments' => Serializer::serialize(['fixture']), + 'payload_codec' => CodecRegistry::defaultCodec(), + 'received_at' => now() + ->toJSON(), + 'accepted_at' => now() + ->toJSON(), + ]]; + if ($kind === 'signal') { + $bundle['history_events'][] = [ + 'id' => (string) Str::ulid(), + 'sequence' => 2, + 'type' => 'SignalWaitOpened', + 'payload' => [ + 'sequence' => 1, + 'signal_name' => 'imported-message', + 'signal_wait_id' => 'import-wait', + ], + 'recorded_at' => now() + ->toJSON(), + ]; + $bundle['tasks'][0]['payload'] = [ + 'resume_source_kind' => 'workflow_signal', + 'workflow_signal_id' => $messageId, + 'signal_name' => 'imported-message', + 'signal_wait_id' => 'import-wait', + ]; + } + + return [$this->resealBundle($bundle), $messageId]; + } + + private function otherMessageCommand(string $kind, ?string $namespace): WorkflowCommand + { + $instance = WorkflowInstance::query()->create([ + 'id' => 'other-' . Str::ulid(), + 'workflow_type' => 'other', + 'workflow_class' => 'OtherWorkflow', + 'namespace' => $namespace, + 'run_count' => 1, + ]); + $run = WorkflowRun::query()->create([ + 'id' => (string) Str::ulid(), + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'workflow_type' => 'other', + 'workflow_class' => 'OtherWorkflow', + 'namespace' => $namespace, + 'status' => 'waiting', + 'payload_codec' => CodecRegistry::defaultCodec(), + ]); + + return WorkflowCommand::query()->create([ + 'id' => (string) Str::ulid(), + 'workflow_instance_id' => $instance->id, + 'workflow_run_id' => $run->id, + 'command_type' => $kind, + 'status' => 'accepted', + 'accepted_at' => now(), + 'message_sequence' => 1, + ])->refresh(); + } + + /** + * @param array $bundle + */ + private function applyImportedMessage(array $bundle, string $kind, string $messageId): void + { + $bridge = app(DefaultWorkflowTaskBridge::class); + $taskId = $bundle['tasks'][0]['id']; + $this->assertTrue($bridge->claimStatus($taskId)['claimed']); + $commands = $kind === 'signal' ? [] : [[ + 'type' => 'complete_update', + 'update_id' => $messageId, + 'result' => Serializer::serialize('completed'), + 'payload_codec' => CodecRegistry::defaultCodec(), + ]]; + $result = $bridge->complete($taskId, $commands); + $this->assertTrue($result['completed'], json_encode($result)); + } + /** * @return array */ From c2d32ac2171587e27f76c6b8ad7393f2e1e62456 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 20:35:48 +0000 Subject: [PATCH 4/6] Keep imported lineage within verified endpoint identities --- docs/workflow/plan.md | 6 + src/V2/Support/ChildRunHistory.php | 41 +- src/V2/Support/EmbeddedV2HistoryImport.php | 55 ++- src/V2/Support/EmbeddedV2ImportContract.php | 5 +- src/V2/Support/EmbeddedV2ImportLineage.php | 138 ++++++ src/V2/Support/ImportedLineageReferences.php | 62 +++ src/V2/Support/RunLineageView.php | 58 ++- .../V2EmbeddedReplayRegressionCorpusTest.php | 50 ++ ...ted-child-lineage-requires-owned-edge.json | 41 ++ .../V2/EmbeddedV2HistoryImportLineageTest.php | 427 ++++++++++++++++++ 10 files changed, 838 insertions(+), 45 deletions(-) create mode 100644 src/V2/Support/EmbeddedV2ImportLineage.php create mode 100644 src/V2/Support/ImportedLineageReferences.php create mode 100644 tests/Fixtures/V2/ReplayRegression/imported-child-lineage-requires-owned-edge.json create mode 100644 tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php diff --git a/docs/workflow/plan.md b/docs/workflow/plan.md index 4f63de995..de9c846aa 100644 --- a/docs/workflow/plan.md +++ b/docs/workflow/plan.md @@ -636,6 +636,12 @@ do not introduce new durable truth. version, payload codec, or compatibility marker the server does not declare support for, and `--dry-run` is the supported rehearsal mode for cutover audits. +- Imported lineage resolves through edges whose run and instance endpoints + exist in the target namespace. Missing endpoints stay in the imported + history without attaching a later unrelated run. Importing the related + bundle can materialize the edge. PostgreSQL write imports must run outside + a caller transaction so the importer can own and retry its serializable + transaction. Preview imports do not write or change transaction isolation. - Cross-database import/export is the same `workflow:v2:history-export` / `workflow:v2:history-import` round trip with the language-neutral `avro` codec; the importer normalizes legacy `workflow-serializer-y` diff --git a/src/V2/Support/ChildRunHistory.php b/src/V2/Support/ChildRunHistory.php index 953840520..474c7449e 100644 --- a/src/V2/Support/ChildRunHistory.php +++ b/src/V2/Support/ChildRunHistory.php @@ -313,14 +313,18 @@ public static function childRunForSequence(WorkflowRun $run, int $sequence): ?Wo $childRun = $childRunId === null ? $link?->childRun - : self::loadRun($childRunId); + : self::loadRun($childRunId, $run); + + if ($childRun instanceof WorkflowRun && ! ImportedLineageReferences::allows($run, $childRun)) { + $childRun = null; + } if ($childRun instanceof WorkflowRun && ($startedEvent !== null || $resolutionEvent !== null)) { return $childRun; } if (! $childRun instanceof WorkflowRun && $childInstanceId !== null) { - $childRun = self::loadCurrentRunForInstance($childInstanceId); + $childRun = self::loadCurrentRunForInstance($childInstanceId, $run); } return self::followContinuedRun($childRun); @@ -481,7 +485,7 @@ public static function childCallIdForRun(WorkflowRun $run): ?string $payload = is_array($startedEvent?->payload) ? $startedEvent->payload : null; $parentRunId = self::stringValue($payload['parent_workflow_run_id'] ?? null); $parentSequence = self::intValue($payload['parent_sequence'] ?? null); - $parentRun = $parentRunId === null ? null : self::loadRun($parentRunId); + $parentRun = $parentRunId === null ? null : self::loadRun($parentRunId, $run); return self::childCallIdFromPayload($payload) ?? self::legacyChildCallIdFromStartedWorkflowPayload($payload) @@ -581,7 +585,10 @@ public static function outputForResolution( WorkflowHistoryEvent $resolutionEvent, ?WorkflowRun $childRun = null, ): mixed { - $childRun ??= self::loadRun(self::stringValue($resolutionEvent->payload['child_workflow_run_id'] ?? null)); + $relatedId = self::stringValue($resolutionEvent->payload['child_workflow_run_id'] ?? null); + if ($childRun === null && $relatedId !== null) { + $childRun = self::loadRun($relatedId, $resolutionEvent->run); + } ExecutionTracing::linkChildOutcome($resolutionEvent, $childRun); $payload = $resolutionEvent->payload['output'] ?? null; @@ -907,31 +914,33 @@ private static function openChildCallForSequence(WorkflowRun $parentRun, int $se return $childCall; } - private static function loadRun(?string $runId): ?WorkflowRun + private static function loadRun(?string $runId, ?WorkflowRun $origin): ?WorkflowRun { - if ($runId === null) { + if ($runId === null || $origin === null) { return null; } - /** @var WorkflowRun|null $run */ - $run = ConfiguredV2Models::query('run_model', WorkflowRun::class) - ->with(['summary', 'instance', 'failures', 'historyEvents']) - ->find($runId); - - return $run; + return ImportedLineageReferences::resolve( + $origin, + $runId, + ['summary', 'instance', 'failures', 'historyEvents'] + ); } - private static function loadCurrentRunForInstance(string $instanceId): ?WorkflowRun + private static function loadCurrentRunForInstance(string $instanceId, WorkflowRun $origin): ?WorkflowRun { /** @var \Workflow\V2\Models\WorkflowInstance|null $instance */ $instance = ConfiguredV2Models::query('instance_model', \Workflow\V2\Models\WorkflowInstance::class) + ->where('namespace', $origin->namespace) ->find($instanceId); if ($instance === null) { return null; } - return CurrentRunResolver::forInstance($instance, ['summary', 'failures', 'historyEvents']); + $run = CurrentRunResolver::forInstance($instance, ['summary', 'failures', 'historyEvents']); + + return $run instanceof WorkflowRun && ImportedLineageReferences::allows($origin, $run) ? $run : null; } private static function followContinuedRun(?WorkflowRun $childRun): ?WorkflowRun @@ -950,6 +959,10 @@ private static function followContinuedRun(?WorkflowRun $childRun): ?WorkflowRun if (! $currentRun instanceof WorkflowRun || $currentRun->id === $childRun->id || isset($visited[$childRun->id])) { return $childRun; } + if ($currentRun->namespace !== $childRun->namespace + || $currentRun->workflow_instance_id !== $childRun->workflow_instance_id) { + return $childRun; + } $visited[$childRun->id] = true; $childRun = $currentRun; diff --git a/src/V2/Support/EmbeddedV2HistoryImport.php b/src/V2/Support/EmbeddedV2HistoryImport.php index c5f2537b7..a84beaceb 100644 --- a/src/V2/Support/EmbeddedV2HistoryImport.php +++ b/src/V2/Support/EmbeddedV2HistoryImport.php @@ -196,13 +196,33 @@ public static function import(array $bundle, array $options = []): array ]; } + if (DB::connection()->getDriverName() === 'pgsql' && DB::transactionLevel() !== 0) { + return [ + ...$base, + 'status' => 'rejected', + 'eligibility' => self::withEligibilityError( + $eligibility, + 'target.transaction_in_progress', + 'PostgreSQL history import must run outside an existing transaction so its serializable transaction can be retried safely.', + ), + 'rows' => self::emptyRows(), + ]; + } + $importId = self::stringValue($options['import_id'] ?? null) ?? (string) Str::ulid(); $importedAt = now(); try { $rows = DB::transaction(static function () use ($bundle, $workflow, $importId, $importedAt): array { + if (DB::connection()->getDriverName() === 'pgsql') { + // Row locks cannot lock a missing endpoint on PostgreSQL. + // Serializable predicate checks prevent reciprocal imports + // from both committing without seeing the other's run. + DB::statement('SET TRANSACTION ISOLATION LEVEL SERIALIZABLE'); + } + return self::writeBundle($bundle, $workflow, $importId, $importedAt); - }); + }, 3); } catch (Throwable $throwable) { return [ ...$base, @@ -236,6 +256,10 @@ private static function writeBundle( string $importId, CarbonInterface $importedAt, ): array { + $lineageErrors = EmbeddedV2ImportLineage::errors($bundle, $workflow, true); + if ($lineageErrors !== []) { + throw new InvalidArgumentException($lineageErrors[0]['message']); + } $rows = self::emptyRows(); $namespace = self::stringValue($workflow['namespace'] ?? null); $runStatus = RunStatus::from($workflow['status']); @@ -339,7 +363,7 @@ private static function writeBundle( $rows['activity_attempts'] += self::writeActivityAttempts($bundle, $runId); $rows['workflow_run_timers'] += self::writeTimers($bundle, $runId); $rows['workflow_failures'] += self::writeFailures($bundle, $runId); - $rows['workflow_links'] += self::writeLinks($bundle, $runId, $instanceId); + $rows['workflow_links'] += self::writeLinks($bundle, $runId, $instanceId, $namespace); $rows['workflow_memos'] += self::writeMemos($workflow, $runId, $instanceId, $historySequence); $rows['workflow_search_attributes'] += self::writeSearchAttributes( $workflow, @@ -727,7 +751,7 @@ private static function writeFailures(array $bundle, string $runId): int return $count; } - private static function writeLinks(array $bundle, string $runId, string $instanceId): int + private static function writeLinks(array $bundle, string $runId, string $instanceId, ?string $namespace): int { $links = self::arrayValue($bundle['links'] ?? null); $count = 0; @@ -739,8 +763,12 @@ private static function writeLinks(array $bundle, string $runId, string $instanc if ($parentInstance === null || $parentRun === null) { continue; } + $related = EmbeddedV2ImportLineage::existingEndpoint($namespace, $parentRun, $parentInstance); + if ($related === null) { + continue; + } - $count += self::insertLink([ + $inserted = self::insertLink([ 'id' => self::ulidValue($link['id'] ?? null) ?? (string) Str::ulid(), 'link_type' => self::stringValue($link['type'] ?? null) ?? 'parent', 'sequence' => self::intValue($link['sequence'] ?? null), @@ -751,7 +779,11 @@ private static function writeLinks(array $bundle, string $runId, string $instanc 'is_primary_parent' => (bool) ($link['is_primary_parent'] ?? false), 'created_at' => self::timestamp($link['created_at'] ?? null) ?? now(), 'updated_at' => self::timestamp($link['created_at'] ?? null) ?? now(), - ]) ? 1 : 0; + ]); + if ($inserted) { + $count++; + RunSummaryProjector::project($related->fresh() ?? $related); + } } foreach (self::listValue($links['children'] ?? null) as $link) { @@ -761,8 +793,12 @@ private static function writeLinks(array $bundle, string $runId, string $instanc if ($childInstance === null || $childRun === null) { continue; } + $related = EmbeddedV2ImportLineage::existingEndpoint($namespace, $childRun, $childInstance); + if ($related === null) { + continue; + } - $count += self::insertLink([ + $inserted = self::insertLink([ 'id' => self::ulidValue($link['id'] ?? null) ?? (string) Str::ulid(), 'link_type' => self::stringValue($link['type'] ?? null) ?? 'child', 'sequence' => self::intValue($link['sequence'] ?? null), @@ -773,7 +809,11 @@ private static function writeLinks(array $bundle, string $runId, string $instanc 'is_primary_parent' => (bool) ($link['is_primary_parent'] ?? false), 'created_at' => self::timestamp($link['created_at'] ?? null) ?? now(), 'updated_at' => self::timestamp($link['created_at'] ?? null) ?? now(), - ]) ? 1 : 0; + ]); + if ($inserted) { + $count++; + RunSummaryProjector::project($related->fresh() ?? $related); + } } return $count; @@ -999,6 +1039,7 @@ private static function eligibility(array $bundle, array $integrity, array $work } $errors = [...$errors, ...EmbeddedV2ImportMessageReferences::errors($bundle)]; + $errors = [...$errors, ...EmbeddedV2ImportLineage::errors($bundle, $workflow)]; return [ 'eligible' => $errors === [], diff --git a/src/V2/Support/EmbeddedV2ImportContract.php b/src/V2/Support/EmbeddedV2ImportContract.php index 4c8590e27..cd44eb937 100644 --- a/src/V2/Support/EmbeddedV2ImportContract.php +++ b/src/V2/Support/EmbeddedV2ImportContract.php @@ -85,12 +85,13 @@ public static function manifest(): array 'activity_tasks' => 'pending activity executions and ready activity tasks remain claimable; running attempts are not importable', 'waits' => 'wait projections are rebuilt from typed history plus imported task/activity/timer rows', 'timers' => 'pending timers are copied with their fire_at timestamp and are visible to timer repair/recovery', - 'lineage' => 'parent and child links from the bundle are copied as diagnostic lineage edges when identifiers are present', + 'lineage' => 'parent and child edges are copied only when both endpoint run/instance pairs exist in the target namespace; missing endpoints remain in history without resolving later unrelated runs, and the counterpart bundle can materialize the edge', ], 'failure_and_rollback' => [ 'transaction' => 'all durable rows and projections are written in one database transaction', + 'postgresql_transaction' => 'write imports own a serializable transaction and reject an existing caller transaction; serialization conflicts retry the complete transaction up to three attempts without changing session defaults', 'partial_or_interrupted_import' => 'no imported rows are committed when the transaction rolls back', - 'retry_same_bundle' => 'returns already_imported when run_id and import_dedupe_key match', + 'retry_same_bundle' => 'returns already_imported when namespace, import source, run_id and import_dedupe_key match', 'conflicting_existing_run' => 'rejected before writing any rows', ], 'visibility_and_audit' => [ diff --git a/src/V2/Support/EmbeddedV2ImportLineage.php b/src/V2/Support/EmbeddedV2ImportLineage.php new file mode 100644 index 000000000..aac673dc0 --- /dev/null +++ b/src/V2/Support/EmbeddedV2ImportLineage.php @@ -0,0 +1,138 @@ + $bundle + * @param array $workflow + * @return list + */ + public static function errors(array $bundle, array $workflow, bool $lock = false): array + { + $errors = []; + foreach (self::references($bundle, $workflow) as [$runId, $instanceId]) { + [, $error] = self::endpoint($workflow['namespace'] ?? null, $runId, $instanceId, $lock); + if ($error !== null) { + $errors[] = $error; + } + } + + return $errors; + } + + public static function existingEndpoint(?string $namespace, string $runId, string $instanceId): ?WorkflowRun + { + [$run, $error] = self::endpoint($namespace, $runId, $instanceId, true); + if ($error !== null) { + throw new InvalidArgumentException($error['message']); + } + + return $run; + } + + /** + * @return array{WorkflowRun|null, array{rule: string, message: string}|null} + */ + private static function endpoint(?string $namespace, string $runId, ?string $instanceId, bool $lock): array + { + $runs = WorkflowRun::query(); + $instances = WorkflowInstance::query(); + if ($lock) { + $runs->lockForUpdate(); + $instances->lockForUpdate(); + } + $run = $runs->find($runId, ['id', 'namespace', 'workflow_instance_id']); + $instance = $instanceId === null ? null : $instances->find($instanceId, ['id', 'namespace']); + if (($run instanceof WorkflowRun && $run->namespace !== $namespace) + || ($instance instanceof WorkflowInstance && $instance->namespace !== $namespace)) { + return [null, [ + 'rule' => 'references.lineage_namespace_conflict', + 'message' => 'A lineage endpoint exists outside the target namespace.', + ]]; + } + if ($run instanceof WorkflowRun && $instanceId !== null + && ($run->workflow_instance_id !== $instanceId || ! $instance instanceof WorkflowInstance)) { + return [null, [ + 'rule' => 'references.lineage_identity_conflict', + 'message' => 'A lineage run does not belong to the declared workflow instance.', + ]]; + } + + return [$run, null]; + } + + /** + * @param array $bundle + * @param array $workflow + * @return list + */ + private static function references(array $bundle, array $workflow): array + { + $references = []; + foreach ([ + 'parents' => 'parent', + 'children' => 'child', + ] as $section => $prefix) { + foreach (self::rows($bundle['links'][$section] ?? null) as $link) { + self::append( + $references, + $link[$prefix . '_workflow_run_id'] ?? null, + $link[$prefix . '_workflow_instance_id'] ?? null + ); + } + } + foreach (self::rows($bundle['history_events'] ?? null) as $event) { + $payload = is_array($event['payload'] ?? null) ? $event['payload'] : []; + if (($event['type'] ?? null) === 'WorkflowStarted') { + self::append( + $references, + $payload['parent_workflow_run_id'] ?? null, + $payload['parent_workflow_instance_id'] ?? null + ); + self::append($references, $payload['continued_from_run_id'] ?? null, $workflow['instance_id'] ?? null); + } elseif (($event['type'] ?? null) === 'WorkflowContinuedAsNew') { + self::append($references, $payload['continued_to_run_id'] ?? null, $workflow['instance_id'] ?? null); + } elseif (in_array( + $event['type'] ?? null, + ['ChildWorkflowScheduled', 'ChildRunStarted', 'ChildRunCompleted', + 'ChildRunFailed', 'ChildRunCancelled', 'ChildRunTerminated', 'ChildRunTimedOut'], + true + )) { + self::append( + $references, + $payload['child_workflow_run_id'] ?? null, + $payload['child_workflow_instance_id'] ?? null + ); + } + } + + return $references; + } + + /** + * @param list $references + */ + private static function append(array &$references, mixed $runId, mixed $instanceId): void + { + if (is_string($runId) && $runId !== '') { + $references[] = [$runId, is_string($instanceId) && $instanceId !== '' ? $instanceId : null]; + } + } + + /** + * @return list> + */ + private static function rows(mixed $value): array + { + return is_array($value) ? array_values(array_filter($value, is_array(...))) : []; + } +} diff --git a/src/V2/Support/ImportedLineageReferences.php b/src/V2/Support/ImportedLineageReferences.php new file mode 100644 index 000000000..377f37629 --- /dev/null +++ b/src/V2/Support/ImportedLineageReferences.php @@ -0,0 +1,62 @@ +import_source === EmbeddedV2ImportContract::IMPORT_SOURCE; + } + + public static function allows(WorkflowRun $origin, WorkflowRun $related): bool + { + if ($origin->namespace !== $related->namespace) { + return false; + } + if ($origin->id === $related->id || ! self::imported($origin)) { + return true; + } + + // Unresolved imported IDs remain history, not authority to attach a + // later run. The importer writes an edge only after locking and + // validating both endpoint run/instance pairs in this namespace. + return ConfiguredV2Models::query('link_model', WorkflowLink::class) + ->where(static function ($query) use ($origin, $related): void { + $query->where(static function ($edge) use ($origin, $related): void { + $edge->where('parent_workflow_run_id', $origin->id) + ->where('parent_workflow_instance_id', $origin->workflow_instance_id) + ->where('child_workflow_run_id', $related->id) + ->where('child_workflow_instance_id', $related->workflow_instance_id); + })->orWhere(static function ($edge) use ($origin, $related): void { + $edge->where('child_workflow_run_id', $origin->id) + ->where('child_workflow_instance_id', $origin->workflow_instance_id) + ->where('parent_workflow_run_id', $related->id) + ->where('parent_workflow_instance_id', $related->workflow_instance_id); + }); + })->exists(); + } + + /** + * @param list $relations + */ + public static function resolve(WorkflowRun $origin, ?string $id, array $relations = []): ?WorkflowRun + { + if ($id === null) { + return null; + } + /** @var WorkflowRun|null $related */ + $related = ConfiguredV2Models::query('run_model', WorkflowRun::class) + ->where('namespace', $origin->namespace) + ->with($relations) + ->find($id); + + return $related instanceof WorkflowRun && self::allows($origin, $related) ? $related : null; + } +} diff --git a/src/V2/Support/RunLineageView.php b/src/V2/Support/RunLineageView.php index 1f44ee9f2..12af4c10e 100644 --- a/src/V2/Support/RunLineageView.php +++ b/src/V2/Support/RunLineageView.php @@ -19,21 +19,22 @@ final class RunLineageView */ public static function parentsForRun(WorkflowRun $run): array { - $run->loadMissing(['historyEvents', 'parentLinks.parentRun.summary', 'instance.runs.summary']); + $run->loadMissing(['historyEvents', 'parentLinks', 'instance.runs.summary']); $entries = []; $seen = []; $continuedFromRunId = ChildRunHistory::continuedFromRunId($run); - if ($continuedFromRunId !== null) { + $continuedFrom = $continuedFromRunId === null ? null : self::resolveRun($run, $continuedFromRunId); + if ($continuedFromRunId !== null && (! ImportedLineageReferences::imported($run) || $continuedFrom !== null)) { $entries[] = self::entry( id: self::stringValue(self::workflowStartedEvent($run)?->payload['workflow_link_id'] ?? null) ?? sprintf('continue_as_new:%s', $continuedFromRunId), linkType: 'continue_as_new', sequence: self::intValue(self::workflowStartedEvent($run)?->payload['sequence'] ?? null), isPrimaryParent: true, - relatedRun: self::resolveRun($run, $continuedFromRunId), + relatedRun: $continuedFrom, instanceId: $run->workflow_instance_id, runId: $continuedFromRunId, createdAt: self::workflowStartedEvent($run)?->recorded_at @@ -45,15 +46,19 @@ public static function parentsForRun(WorkflowRun $run): array } $parentReference = ChildRunHistory::parentReferenceForRun($run); + $parentRun = $parentReference === null ? null : self::resolveRun( + $run, + $parentReference['parent_workflow_run_id'] + ); - if ($parentReference !== null) { + if ($parentReference !== null && (! ImportedLineageReferences::imported($run) || $parentRun !== null)) { $entries[] = self::entry( id: $parentReference['child_call_id'] ?? sprintf('child_workflow:%s:%s', $parentReference['parent_workflow_run_id'], $run->id), linkType: 'child_workflow', sequence: $parentReference['parent_sequence'], isPrimaryParent: true, - relatedRun: self::resolveRun($run, $parentReference['parent_workflow_run_id']), + relatedRun: $parentRun, instanceId: $parentReference['parent_workflow_instance_id'], runId: $parentReference['parent_workflow_run_id'], relationPrefix: 'parent', @@ -76,6 +81,12 @@ public static function parentsForRun(WorkflowRun $run): array if (isset($seen[$key])) { continue; } + $related = self::resolveRun($run, $link->parent_workflow_run_id); + if (($related === null && ImportedLineageReferences::imported($run)) + || ($related !== null && $related->workflow_instance_id !== $link->parent_workflow_instance_id) + || $link->child_workflow_instance_id !== $run->workflow_instance_id) { + continue; + } $entries[] = self::entry( id: $link->link_type === 'child_workflow' @@ -84,7 +95,7 @@ public static function parentsForRun(WorkflowRun $run): array linkType: $link->link_type, sequence: $link->sequence, isPrimaryParent: $link->is_primary_parent, - relatedRun: $link->parentRun, + relatedRun: $related, instanceId: $link->parent_workflow_instance_id, runId: $link->parent_workflow_run_id, relationPrefix: 'parent', @@ -107,12 +118,7 @@ public static function parentsForRun(WorkflowRun $run): array */ public static function continuedWorkflowsForRun(WorkflowRun $run): array { - $run->loadMissing([ - 'historyEvents', - 'childLinks.childRun.summary', - 'childLinks.childRun.instance.currentRun.summary', - 'instance.runs.summary', - ]); + $run->loadMissing(['historyEvents', 'childLinks', 'instance.runs.summary']); $entries = []; $seen = []; @@ -123,15 +129,16 @@ public static function continuedWorkflowsForRun(WorkflowRun $run): array if ($continuedEvent instanceof WorkflowHistoryEvent) { $continuedRunId = self::stringValue($continuedEvent->payload['continued_to_run_id'] ?? null); + $continuedRun = $continuedRunId === null ? null : self::resolveRun($run, $continuedRunId); - if ($continuedRunId !== null) { + if ($continuedRunId !== null && (! ImportedLineageReferences::imported($run) || $continuedRun !== null)) { $entries[] = self::entry( id: self::stringValue($continuedEvent->payload['workflow_link_id'] ?? null) ?? sprintf('continue_as_new:%s', $continuedRunId), linkType: 'continue_as_new', sequence: self::intValue($continuedEvent->payload['sequence'] ?? null), isPrimaryParent: true, - relatedRun: self::resolveRun($run, $continuedRunId), + relatedRun: $continuedRun, instanceId: $run->workflow_instance_id, runId: $continuedRunId, relationPrefix: 'child', @@ -150,6 +157,9 @@ public static function continuedWorkflowsForRun(WorkflowRun $run): array foreach (self::childSequences($run) as $sequence) { $childRun = ChildRunHistory::childRunForSequence($run, $sequence); + if ($childRun === null && ImportedLineageReferences::imported($run)) { + continue; + } $link = ChildRunHistory::latestLinkForSequence($run, $sequence); $startedEvent = ChildRunHistory::startedEventForSequence($run, $sequence); $scheduledEvent = ChildRunHistory::scheduledEventForSequence($run, $sequence); @@ -217,6 +227,12 @@ public static function continuedWorkflowsForRun(WorkflowRun $run): array if (! $link instanceof WorkflowLink) { continue; } + $related = self::resolveRun($run, $link->child_workflow_run_id); + if (($related === null && ImportedLineageReferences::imported($run)) + || ($related !== null && $related->workflow_instance_id !== $link->child_workflow_instance_id) + || $link->parent_workflow_instance_id !== $run->workflow_instance_id) { + continue; + } $childCallId = $link->link_type === 'child_workflow' && $link->sequence !== null ? ChildRunHistory::childCallIdForSequence($run, (int) $link->sequence) @@ -244,7 +260,7 @@ public static function continuedWorkflowsForRun(WorkflowRun $run): array linkType: $link->link_type, sequence: $link->sequence, isPrimaryParent: $link->is_primary_parent, - relatedRun: $link->childRun, + relatedRun: $related, instanceId: $link->child_workflow_instance_id, runId: $link->child_workflow_run_id, relationPrefix: 'child', @@ -325,18 +341,16 @@ private static function resolveRun(WorkflowRun $selectedRun, string $runId): ?Wo $sameInstanceRun = $selectedRun->instance?->runs?->firstWhere('id', $runId); - if ($sameInstanceRun instanceof WorkflowRun) { + if ($sameInstanceRun instanceof WorkflowRun && ImportedLineageReferences::allows( + $selectedRun, + $sameInstanceRun + )) { $sameInstanceRun->loadMissing('summary'); return $sameInstanceRun; } - /** @var WorkflowRun|null $run */ - $run = ConfiguredV2Models::query('run_model', \Workflow\V2\Models\WorkflowRun::class) - ->with('summary') - ->find($runId); - - return $run; + return ImportedLineageReferences::resolve($selectedRun, $runId, ['summary']); } /** diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index b2e788402..7f46ade96 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -58,6 +58,7 @@ use Workflow\V2\Support\CancellationScopeDelivery; use Workflow\V2\Support\CancellationScopeHistory; use Workflow\V2\Support\CancellationScopeRequests; +use Workflow\V2\Support\ChildRunHistory; use Workflow\V2\Support\ConditionWaits; use Workflow\V2\Support\DefaultHistoryProjectionRole; use Workflow\V2\Support\DefaultWorkflowTaskBridge; @@ -66,6 +67,7 @@ use Workflow\V2\Support\QueryStateReplayer; use Workflow\V2\Support\ReplayDiff; use Workflow\V2\Support\RunActivityView; +use Workflow\V2\Support\RunDetailView; use Workflow\V2\Support\RunTimelineProjector; use Workflow\V2\Support\WorkflowCommandNormalizer; use Workflow\V2\Support\WorkflowFiberRunner; @@ -85,6 +87,54 @@ final class V2EmbeddedReplayRegressionCorpusTest extends TestCase private int $workflowNumber = 0; + public function testImportedChildLineageCorpusDoesNotAttachAnUnrelatedLaterRun(): void + { + $path = self::FIXTURE_DIR . '/imported-child-lineage-requires-owned-edge.json'; + if (! is_file($path)) { + $this->markTestSkipped('The retained imported-lineage evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + self::stopWorkers(); + Queue::fake(); + config([ + 'queue.default' => 'database', + ]); + $this->clearWorkflowState(); + $parent = $this->createRunFromFixture($fixture); + $bundle = HistoryExport::forRun($parent->fresh()); + $this->assertCount(1, $bundle['links']['children']); + $this->clearWorkflowState(); + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundle, [ + 'namespace' => 'lineage-import', + ])['status']); + $this->assertSame(0, WorkflowLink::query()->count()); + $identity = $fixture['history'][2]['payload']; + WorkflowInstance::query()->create([ + 'id' => $identity['child_workflow_instance_id'], + 'namespace' => 'lineage-import', + 'workflow_type' => 'unrelated-later', + 'workflow_class' => 'unrelated-later', + 'current_run_id' => $identity['child_workflow_run_id'], + 'run_count' => 1, + ]); + WorkflowRun::query()->create([ + 'id' => $identity['child_workflow_run_id'], + 'workflow_instance_id' => $identity['child_workflow_instance_id'], + 'namespace' => 'lineage-import', + 'workflow_type' => 'unrelated-later', + 'workflow_class' => 'unrelated-later', + 'run_number' => 1, + 'status' => RunStatus::Waiting, + 'payload_codec' => $fixture['workflow']['payload_codec'], + 'arguments' => Serializer::serialize([]), + ]); + $imported = WorkflowRun::query()->findOrFail($parent->id); + $this->assertNull(ChildRunHistory::childRunForSequence($imported, 1)); + foreach ([RunDetailView::forRun($imported->fresh()), HistoryExport::forRun($imported->fresh())] as $view) { + $this->assertStringNotContainsString('unrelated-later', json_encode($view, JSON_THROW_ON_ERROR)); + } + } + public function testRetriedActivityCancellationCorpusKeepsTheCompletedTaskAndAcknowledgesTheLiveTask(): void { $path = self::FIXTURE_DIR . '/retried-activity-cancellation-live-task.json'; diff --git a/tests/Fixtures/V2/ReplayRegression/imported-child-lineage-requires-owned-edge.json b/tests/Fixtures/V2/ReplayRegression/imported-child-lineage-requires-owned-edge.json new file mode 100644 index 000000000..bab79e7bc --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/imported-child-lineage-requires-owned-edge.json @@ -0,0 +1,41 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "imported-child-lineage-requires-owned-edge", + "protocol_version": "1.0", + "bindings": ["php"], + "consumers": ["workflow-fiber-runner", "embedded-history-import"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestParentWaitingOnChildWorkflow", + "arguments": [5], + "payload_codec": "avro" + }, + "history": [ + {"sequence": 1, "event_type": "WorkflowStarted", "payload": {}, "recorded_at": "2026-10-25T12:00:00Z"}, + { + "sequence": 2, + "event_type": "ChildWorkflowScheduled", + "payload": { + "sequence": 1, + "child_workflow_type": "Tests\\Fixtures\\V2\\TestTimerWorkflow", + "child_workflow_instance_id": "imported-lineage-child", + "child_workflow_run_id": "01J00000000000000000000001", + "child_call_id": "01J00000000000000000000002" + }, + "recorded_at": "2026-10-25T12:00:01Z" + }, + { + "sequence": 3, + "event_type": "ChildRunStarted", + "payload": { + "sequence": 1, + "child_workflow_instance_id": "imported-lineage-child", + "child_workflow_run_id": "01J00000000000000000000001", + "child_call_id": "01J00000000000000000000002" + }, + "recorded_at": "2026-10-25T12:00:02Z" + } + ], + "history_import_metadata": {"memo": {}, "search_attributes": {}}, + "expected": {"completed": false, "result": null, "commands": []} +} diff --git a/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php b/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php new file mode 100644 index 000000000..4c53d2b0c --- /dev/null +++ b/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php @@ -0,0 +1,427 @@ +sourcePair(); + $this->clearState(); + $other = $selected === 'parent' ? 'child' : 'parent'; + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundles[$other], [ + 'namespace' => $foreign, + ])['status']); + $before = $this->snapshot(); + + $report = EmbeddedV2HistoryImport::import($bundles[$selected], [ + 'namespace' => 'tenant-a', + 'dry_run' => $dryRun, + ]); + + $this->assertSame('rejected', $report['status']); + $this->assertContains( + 'references.lineage_namespace_conflict', + array_column($report['eligibility']['errors'], 'rule') + ); + $this->assertSame(0, array_sum($report['rows'])); + $this->assertSame($before, $this->snapshot()); + } + + public static function foreignEndpoints(): iterable + { + foreach (['parent', 'child'] as $selected) { + foreach (['tenant-b', null] as $foreign) { + foreach ([false, true] as $dryRun) { + yield $selected . '/' . ($foreign ?? 'null') . ($dryRun ? '/preview' : '/write') => [ + $selected, + $foreign, + $dryRun, + ]; + } + } + } + } + + #[DataProvider('importOrders')] + public function testReciprocalBundlesMaterializeOneScopedEdgeInEitherOrder(string $first, ?string $namespace): void + { + $bundles = $this->sourcePair(); + $this->clearState(); + $second = $first === 'parent' ? 'child' : 'parent'; + + $initial = EmbeddedV2HistoryImport::import($bundles[$first], [ + 'namespace' => $namespace, + ]); + + $this->assertSame('imported', $initial['status']); + $this->assertSame(0, $initial['rows']['workflow_links']); + $this->assertSame(0, WorkflowLink::query()->count()); + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundles[$second], [ + 'namespace' => $namespace, + ])['status']); + $link = WorkflowLink::query()->sole(); + $this->assertSame($bundles['parent']['workflow']['run_id'], $link->parent_workflow_run_id); + $this->assertSame($bundles['child']['workflow']['run_id'], $link->child_workflow_run_id); + foreach ($bundles as $name => $bundle) { + $run = WorkflowRun::query()->findOrFail($bundle['workflow']['run_id']); + $this->assertSame($namespace, $run->namespace); + $export = HistoryExport::forRun($run); + $entries = $export['links'][$name === 'parent' ? 'children' : 'parents']; + $this->assertCount(1, $entries); + $this->assertSame('waiting', $entries[0]['status']); + $this->assertSame('already_imported', EmbeddedV2HistoryImport::import($bundle, [ + 'namespace' => $namespace, + ])['status']); + } + $this->assertSame(1, WorkflowLink::query()->count()); + } + + public static function importOrders(): iterable + { + foreach (['parent', 'child'] as $first) { + foreach (['tenant-a', null] as $namespace) { + yield $first . '/' . ($namespace ?? 'null') => [$first, $namespace]; + } + } + } + + public function testConcurrentPostgresReciprocalImportsKeepTheirScopedEdge(): void + { + if (DB::connection()->getDriverName() !== 'pgsql' || ! function_exists('pcntl_fork') + || ! function_exists('posix_kill') || ! function_exists('stream_socket_pair')) { + $this->markTestSkipped('PostgreSQL and process control are required for this serialization race.'); + } + $bundles = $this->sourcePair(); + $this->clearState(); + $defaultIsolation = DB::selectOne('SHOW transaction_isolation')->transaction_isolation; + DB::purge(); + $actors = []; + $prepared = []; + $attempts = []; + try { + foreach ($bundles as $name => $bundle) { + $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + $this->assertIsArray($sockets); + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid); + if ($pid === 0) { + fclose($sockets[0]); + stream_set_timeout($sockets[1], 15); + try { + DB::reconnect(); + $paused = false; + $beginnings = 0; + $this->app['events']->listen(TransactionBeginning::class, static function () use ( + &$beginnings + ): void { + $beginnings++; + }); + $this->app['events']->listen(TransactionCommitting::class, static function () use ( + &$paused, + $sockets + ): void { + if ($paused || DB::transactionLevel() !== 1) { + return; + } + $paused = true; + fwrite($sockets[1], json_encode([ + 'connection' => (int) DB::selectOne('SELECT pg_backend_pid() AS id')->id, + 'runs' => WorkflowRun::query()->count(), + 'links' => WorkflowLink::query()->count(), + ], JSON_THROW_ON_ERROR) . "\n"); + if (fgets($sockets[1]) !== "commit\n") { + throw new RuntimeException('Import commit barrier was not released.'); + } + }); + $report = EmbeddedV2HistoryImport::import($bundle, [ + 'namespace' => 'tenant-a', + ]); + fwrite($sockets[1], json_encode([ + 'status' => $report['status'], + 'attempts' => $beginnings, + 'errors' => $report['eligibility']['errors'], + 'transaction_level' => DB::transactionLevel(), + 'isolation' => DB::selectOne('SHOW transaction_isolation')->transaction_isolation, + ], JSON_THROW_ON_ERROR) . "\n"); + DB::disconnect(); + fclose($sockets[1]); + exit(0); + } catch (Throwable $error) { + fwrite($sockets[1], json_encode([ + 'error' => $error->getMessage(), + ]) . "\n"); + exit(1); + } + } + fclose($sockets[1]); + stream_set_timeout($sockets[0], 15); + $actors[$name] = [ + 'pid' => $pid, + 'socket' => $sockets[0], + ]; + $message = fgets($sockets[0]); + $this->assertIsString($message); + $prepared[$name] = json_decode($message, true, flags: JSON_THROW_ON_ERROR); + $this->assertSame(1, $prepared[$name]['runs'] ?? null, $message); + $this->assertSame(0, $prepared[$name]['links'] ?? null, $message); + } + $this->assertNotSame($prepared['parent']['connection'], $prepared['child']['connection']); + DB::reconnect(); + $this->assertSame(0, WorkflowRun::query()->count(), 'Both import transactions are still uncommitted.'); + foreach ($actors as $actor) { + fwrite($actor['socket'], "commit\n"); + } + foreach ($actors as $name => $actor) { + $message = fgets($actor['socket']); + $this->assertIsString($message); + $report = json_decode($message, true, flags: JSON_THROW_ON_ERROR); + $this->assertSame('imported', $report['status'] ?? null, $message); + $this->assertSame(0, $report['transaction_level']); + $this->assertSame($defaultIsolation, $report['isolation']); + $attempts[$name] = $report['attempts']; + pcntl_waitpid($actor['pid'], $status); + $this->assertTrue(pcntl_wifexited($status)); + $this->assertSame(0, pcntl_wexitstatus($status)); + fclose($actor['socket']); + unset($actors[$name]); + } + $this->assertGreaterThan( + 2, + array_sum($attempts), + 'A serialization conflict must retry a complete import transaction.' + ); + $this->assertSame(2, WorkflowRun::query()->count()); + $link = WorkflowLink::query()->sole(); + $this->assertSame($bundles['parent']['workflow']['run_id'], $link->parent_workflow_run_id); + $this->assertSame($bundles['child']['workflow']['run_id'], $link->child_workflow_run_id); + foreach ($bundles as $bundle) { + $this->assertSame('already_imported', EmbeddedV2HistoryImport::import($bundle, [ + 'namespace' => 'tenant-a', + ])['status']); + } + $this->assertSame(1, WorkflowLink::query()->count()); + } finally { + foreach ($actors as $actor) { + posix_kill($actor['pid'], SIGKILL); + pcntl_waitpid($actor['pid'], $status); + fclose($actor['socket']); + } + DB::purge(); + DB::reconnect(); + } + } + + public function testPostgresWriteRefusesAnExistingTransactionWithoutAbortingIt(): void + { + if (DB::connection()->getDriverName() !== 'pgsql') { + $this->markTestSkipped('The PostgreSQL transaction isolation boundary is required.'); + } + $bundle = $this->sourcePair()['parent']; + $this->clearState(); + DB::beginTransaction(); + try { + $this->createRun('outer-owned', (string) Str::ulid(), 'tenant-a', 'outer-owned'); + $before = $this->snapshot(); + $report = EmbeddedV2HistoryImport::import($bundle, [ + 'namespace' => 'tenant-a', + ]); + $this->assertSame('rejected', $report['status']); + $this->assertContains( + 'target.transaction_in_progress', + array_column($report['eligibility']['errors'], 'rule') + ); + $this->assertSame(0, array_sum($report['rows'])); + $this->assertSame(1, DB::transactionLevel()); + $this->assertTrue(DB::connection()->getPdo()->inTransaction()); + $this->assertSame($before, $this->snapshot()); + $this->assertSame('dry_run', EmbeddedV2HistoryImport::import($bundle, [ + 'namespace' => 'tenant-a', + 'dry_run' => true, + ])['status']); + } finally { + DB::rollBack(); + } + $this->assertSame(0, WorkflowRun::query()->count()); + } + + #[DataProvider('lateEndpoints')] + public function testOmittedEdgeCannotCaptureAnUnrelatedLaterRun(string $selected, string $namespace): void + { + $bundles = $this->sourcePair(); + $this->clearState(); + $other = $selected === 'parent' ? 'child' : 'parent'; + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundles[$selected], [ + 'namespace' => 'tenant-a', + ])['status']); + // Isolate the proposed edge-omission path from the old physical writer. + // Typed history must not bypass the absence of a verified edge. + WorkflowLink::query()->delete(); + $this->assertSame(0, WorkflowLink::query()->count()); + $opposite = $bundles[$other]['workflow']; + $unrelated = $this->createRun($opposite['instance_id'], $opposite['run_id'], $namespace, 'unrelated-later'); + $before = $unrelated->getAttributes(); + $run = WorkflowRun::query()->findOrFail($bundles[$selected]['workflow']['run_id']); + + if ($selected === 'parent') { + $this->assertNull(ChildRunHistory::childRunForSequence($run, 1)); + } + $entries = $selected === 'parent' ? RunLineageView::continuedWorkflowsForRun( + $run + ) : RunLineageView::parentsForRun($run); + $this->assertStringNotContainsString('unrelated-later', json_encode($entries, JSON_THROW_ON_ERROR)); + RunLineageProjector::project($run->fresh()); + foreach ([RunDetailView::forRun($run->fresh()), HistoryExport::forRun($run->fresh())] as $view) { + $this->assertStringNotContainsString('unrelated-later', json_encode($view, JSON_THROW_ON_ERROR)); + } + $this->assertSame(0, WorkflowLink::query()->count()); + $this->assertSame($before, $unrelated->fresh()->getAttributes()); + } + + public static function lateEndpoints(): iterable + { + foreach (['parent', 'child'] as $selected) { + foreach (['tenant-a', 'tenant-b'] as $namespace) { + yield $selected . '/' . $namespace => [$selected, $namespace]; + } + } + } + + /** + * @return array{parent: array, child: array} + */ + private function sourcePair(): array + { + $fixture = json_decode( + file_get_contents( + __DIR__ . '/../../Fixtures/V2/ReplayRegression/imported-child-lineage-requires-owned-edge.json' + ), + true, + flags: JSON_THROW_ON_ERROR + ); + $identity = $fixture['history'][2]['payload']; + $parent = $this->createRun( + 'parent-' . Str::ulid(), + (string) Str::ulid(), + 'source', + $fixture['workflow']['type'] + ); + $parent->forceFill([ + 'arguments' => Serializer::serialize($fixture['workflow']['arguments']), + ])->save(); + $child = $this->createRun( + $identity['child_workflow_instance_id'], + $identity['child_workflow_run_id'], + 'source', + $fixture['history'][1]['payload']['child_workflow_type'] + ); + $linkId = $identity['child_call_id']; + foreach ($fixture['history'] as $event) { + WorkflowHistoryEvent::record($parent, HistoryEventType::from($event['event_type']), $event['payload']); + } + WorkflowHistoryEvent::record($child, HistoryEventType::WorkflowStarted, [ + 'parent_workflow_instance_id' => $parent->workflow_instance_id, + 'parent_workflow_run_id' => $parent->id, + 'parent_sequence' => 1, + 'child_call_id' => $linkId, + ]); + WorkflowLink::query()->create([ + 'id' => $linkId, + 'link_type' => 'child_workflow', + 'sequence' => 1, + 'parent_workflow_instance_id' => $parent->workflow_instance_id, + 'parent_workflow_run_id' => $parent->id, + 'child_workflow_instance_id' => $child->workflow_instance_id, + 'child_workflow_run_id' => $child->id, + 'is_primary_parent' => true, + ]); + $bundles = [ + 'parent' => HistoryExport::forRun($parent->fresh()), + 'child' => HistoryExport::forRun($child->fresh()), + ]; + $this->assertCount(1, $bundles['parent']['links']['children']); + $this->assertCount(1, $bundles['child']['links']['parents']); + + return $bundles; + } + + private function createRun(string $instanceId, string $runId, ?string $namespace, string $type): WorkflowRun + { + WorkflowInstance::query()->create([ + 'id' => $instanceId, + 'workflow_type' => $type, + 'workflow_class' => $type, + 'namespace' => $namespace, + 'current_run_id' => $runId, + 'run_count' => 1, + ]); + + return WorkflowRun::query()->create([ + 'id' => $runId, + 'workflow_instance_id' => $instanceId, + 'run_number' => 1, + 'namespace' => $namespace, + 'workflow_type' => $type, + 'workflow_class' => $type, + 'status' => 'waiting', + 'payload_codec' => CodecRegistry::defaultCodec(), + 'arguments' => Serializer::serialize([]), + ])->refresh(); + } + + private function clearState(): void + { + foreach (self::TABLES as $table) { + DB::table($table)->delete(); + } + } + + /** + * @return array + */ + private function snapshot(): array + { + $result = []; + foreach (self::TABLES as $table) { + $result[$table] = DB::table($table)->get()->map(static fn (object $row): array => (array) $row)->all(); + } + + return $result; + } +} From 5e01a938a8766af8f47861aba69d5497bf6731d0 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 21:45:49 +0000 Subject: [PATCH 5/6] Fence imported child effects and reject self lineage --- src/V2/Support/CancellationScopeDelivery.php | 3 + src/V2/Support/EmbeddedV2HistoryImport.php | 4 +- src/V2/Support/EmbeddedV2ImportLineage.php | 18 +- src/V2/Support/ImportedLineageReferences.php | 76 ++++- src/V2/Support/ScopedChildCancellation.php | 9 + .../ScopedChildCancellationDelivery.php | 9 +- src/V2/Support/WorkflowExecutor.php | 26 +- src/V2/WorkflowStub.php | 37 ++- .../V2/V2ScopedChildCancellationTest.php | 228 +++++++++++++++ .../V2/EmbeddedV2HistoryImportLineageTest.php | 270 ++++++++++++++++++ 10 files changed, 657 insertions(+), 23 deletions(-) diff --git a/src/V2/Support/CancellationScopeDelivery.php b/src/V2/Support/CancellationScopeDelivery.php index 96be81dfe..edf93c9fe 100644 --- a/src/V2/Support/CancellationScopeDelivery.php +++ b/src/V2/Support/CancellationScopeDelivery.php @@ -667,6 +667,9 @@ private static function readBoundary(WorkflowRun $run, string $scopeId, bool $pr ); } } catch (LogicException $error) { + if ($error->getMessage() === 'cancellation_scope_child_target_mismatch') { + throw $error; + } throw new LogicException('cancellation_scope_delivery_history_invalid', previous: $error); } return $event; diff --git a/src/V2/Support/EmbeddedV2HistoryImport.php b/src/V2/Support/EmbeddedV2HistoryImport.php index a84beaceb..6bd81c968 100644 --- a/src/V2/Support/EmbeddedV2HistoryImport.php +++ b/src/V2/Support/EmbeddedV2HistoryImport.php @@ -763,7 +763,7 @@ private static function writeLinks(array $bundle, string $runId, string $instanc if ($parentInstance === null || $parentRun === null) { continue; } - $related = EmbeddedV2ImportLineage::existingEndpoint($namespace, $parentRun, $parentInstance); + $related = EmbeddedV2ImportLineage::existingEndpoint($namespace, $parentRun, $parentInstance, $runId); if ($related === null) { continue; } @@ -793,7 +793,7 @@ private static function writeLinks(array $bundle, string $runId, string $instanc if ($childInstance === null || $childRun === null) { continue; } - $related = EmbeddedV2ImportLineage::existingEndpoint($namespace, $childRun, $childInstance); + $related = EmbeddedV2ImportLineage::existingEndpoint($namespace, $childRun, $childInstance, $runId); if ($related === null) { continue; } diff --git a/src/V2/Support/EmbeddedV2ImportLineage.php b/src/V2/Support/EmbeddedV2ImportLineage.php index aac673dc0..ed1b2e051 100644 --- a/src/V2/Support/EmbeddedV2ImportLineage.php +++ b/src/V2/Support/EmbeddedV2ImportLineage.php @@ -20,6 +20,13 @@ public static function errors(array $bundle, array $workflow, bool $lock = false { $errors = []; foreach (self::references($bundle, $workflow) as [$runId, $instanceId]) { + if ($runId === ($workflow['run_id'] ?? null)) { + $errors[] = [ + 'rule' => 'references.lineage_self_reference', + 'message' => 'A workflow run cannot be its own parent, child or continuation.', + ]; + continue; + } [, $error] = self::endpoint($workflow['namespace'] ?? null, $runId, $instanceId, $lock); if ($error !== null) { $errors[] = $error; @@ -29,8 +36,15 @@ public static function errors(array $bundle, array $workflow, bool $lock = false return $errors; } - public static function existingEndpoint(?string $namespace, string $runId, string $instanceId): ?WorkflowRun - { + public static function existingEndpoint( + ?string $namespace, + string $runId, + string $instanceId, + string $originRunId + ): ?WorkflowRun { + if ($runId === $originRunId) { + throw new InvalidArgumentException('A workflow run cannot be its own parent, child or continuation.'); + } [$run, $error] = self::endpoint($namespace, $runId, $instanceId, true); if ($error !== null) { throw new InvalidArgumentException($error['message']); diff --git a/src/V2/Support/ImportedLineageReferences.php b/src/V2/Support/ImportedLineageReferences.php index 377f37629..68ad6088e 100644 --- a/src/V2/Support/ImportedLineageReferences.php +++ b/src/V2/Support/ImportedLineageReferences.php @@ -4,6 +4,8 @@ namespace Workflow\V2\Support; +use Workflow\V2\Enums\HistoryEventType; +use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowLink; use Workflow\V2\Models\WorkflowRun; @@ -17,10 +19,10 @@ public static function imported(WorkflowRun $run): bool public static function allows(WorkflowRun $origin, WorkflowRun $related): bool { - if ($origin->namespace !== $related->namespace) { + if ($origin->namespace !== $related->namespace || $origin->id === $related->id) { return false; } - if ($origin->id === $related->id || ! self::imported($origin)) { + if (! self::imported($origin)) { return true; } @@ -43,6 +45,76 @@ public static function allows(WorkflowRun $origin, WorkflowRun $related): bool })->exists(); } + /** + * Imported child effects require the directed edge and its typed call identity. + * Mutating callers retain their endpoint locks and request the edge/history locks. + */ + public static function allowsChild( + WorkflowRun $parent, + WorkflowRun $child, + ?int $sequence, + ?string $callId = null, + bool $lock = false, + ): bool { + if ($parent->namespace !== $child->namespace || $parent->id === $child->id) { + return false; + } + if (! self::imported($parent) && ! self::imported($child)) { + return true; + } + if ($sequence === null || $sequence < 1) { + return false; + } + $edges = ConfiguredV2Models::query('link_model', WorkflowLink::class) + ->where('link_type', 'child_workflow') + ->where('sequence', $sequence) + ->where('parent_workflow_run_id', $parent->id) + ->where('parent_workflow_instance_id', $parent->workflow_instance_id) + ->where('child_workflow_run_id', $child->id) + ->where('child_workflow_instance_id', $child->workflow_instance_id); + if ($lock) { + $edges->lockForUpdate(); + } + $links = $edges->get(); + if ($links->isEmpty()) { + return false; + } + $history = $parent->historyEvents() + ->whereIn('event_type', [ + HistoryEventType::ChildWorkflowScheduled->value, HistoryEventType::ChildRunStarted->value, + ])->where('payload->sequence', $sequence); + if ($lock) { + $history->lockForUpdate(); + } + foreach ($history->get() as $event) { + if (! $event instanceof WorkflowHistoryEvent) { + continue; + } + $payload = $event->payload; + $recordedCall = $payload['child_call_id'] ?? $payload['workflow_link_id'] ?? null; + if (($payload['sequence'] ?? null) !== $sequence + || ($payload['child_workflow_run_id'] ?? null) !== $child->id + || ($payload['child_workflow_instance_id'] ?? null) !== $child->workflow_instance_id + || ! is_string($recordedCall) || $recordedCall === '' + || ($callId !== null && $recordedCall !== $callId)) { + continue; + } + foreach ($links as $link) { + // Exported logical child edges retain the call ID; a live + // retry/continuation edge can instead have its own link ID. + if ($link->parent_workflow_run_id === $parent->id + && $link->parent_workflow_instance_id === $parent->workflow_instance_id + && $link->child_workflow_run_id === $child->id + && $link->child_workflow_instance_id === $child->workflow_instance_id + && ($link->id === $recordedCall || $link->id === ($payload['workflow_link_id'] ?? null))) { + return true; + } + } + } + + return false; + } + /** * @param list $relations */ diff --git a/src/V2/Support/ScopedChildCancellation.php b/src/V2/Support/ScopedChildCancellation.php index 7e610c030..36fe2c48c 100644 --- a/src/V2/Support/ScopedChildCancellation.php +++ b/src/V2/Support/ScopedChildCancellation.php @@ -68,6 +68,15 @@ public static function members(WorkflowRun $run, string $scopeId): array } $runId = $next['child_workflow_run_id']; } + if (ImportedLineageReferences::imported($run)) { + $child = ConfiguredV2Models::query('run_model', WorkflowRun::class) + ->where('namespace', $run->namespace) + ->find($runId); + if (! $child instanceof WorkflowRun + || ! ImportedLineageReferences::allowsChild($run, $child, $sequence, $callId)) { + throw new LogicException('cancellation_scope_child_target_mismatch'); + } + } $calls[$callId] = true; $sequences[$sequence] = true; $members[] = [ diff --git a/src/V2/Support/ScopedChildCancellationDelivery.php b/src/V2/Support/ScopedChildCancellationDelivery.php index 8bcce1dce..797f21ae5 100644 --- a/src/V2/Support/ScopedChildCancellationDelivery.php +++ b/src/V2/Support/ScopedChildCancellationDelivery.php @@ -96,7 +96,14 @@ public static function request( $member['child_workflow_run_id'] ); if ($child === null || $child->workflow_instance_id !== $member['child_workflow_instance_id'] - || $child->namespace !== $parent->namespace) { + || $child->namespace !== $parent->namespace + || ! ImportedLineageReferences::allowsChild( + $parent, + $child, + $member['sequence'], + $childCallId, + true + )) { throw new LogicException('cancellation_scope_child_target_mismatch'); } $base = self::base($parent, $preparation, $member); diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index 5a5ab5882..f2b2faeab 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -5003,7 +5003,7 @@ private function dispatchParentResumeTasks(WorkflowRun $childRun): void ->lockForUpdate() ->get(); - /** @var array $parentReferences */ + /** @var array $parentReferences */ $parentReferences = []; foreach ($parentLinks as $parentLink) { @@ -5014,13 +5014,14 @@ private function dispatchParentResumeTasks(WorkflowRun $childRun): void $parentSequence = self::intValue($parentLink->sequence); $parentReferences[$parentLink->parent_workflow_run_id] = [ + 'parent_workflow_instance_id' => $parentLink->parent_workflow_instance_id, 'parent_workflow_run_id' => $parentLink->parent_workflow_run_id, 'parent_sequence' => $parentSequence ?? ($parentReferences[$parentLink->parent_workflow_run_id]['parent_sequence'] ?? null), ]; } - if ($parentReferences === []) { + if ($parentReferences === [] && ! ImportedLineageReferences::imported($childRun)) { $parentReference = ChildRunHistory::parentReferenceForRun($childRun); if ($parentReference !== null) { @@ -5031,15 +5032,24 @@ private function dispatchParentResumeTasks(WorkflowRun $childRun): void foreach ($parentReferences as $parentReference) { /** @var WorkflowRun|null $parentRun */ $parentRun = WorkflowRun::query() + ->where('namespace', $childRun->namespace) ->lockForUpdate() ->find($parentReference['parent_workflow_run_id']); - if ($parentRun === null || in_array($parentRun->status, [ - RunStatus::Completed, - RunStatus::Failed, - RunStatus::Cancelled, - RunStatus::Terminated, - ], true)) { + if ($parentRun === null || $parentRun->workflow_instance_id !== $parentReference['parent_workflow_instance_id'] + || ! ImportedLineageReferences::allowsChild( + $parentRun, + $childRun, + $parentReference['parent_sequence'], + $parentReference['child_call_id'] ?? null, + true + ) + || in_array($parentRun->status, [ + RunStatus::Completed, + RunStatus::Failed, + RunStatus::Cancelled, + RunStatus::Terminated, + ], true)) { continue; } diff --git a/src/V2/WorkflowStub.php b/src/V2/WorkflowStub.php index 04c1a54ce..033b1a872 100644 --- a/src/V2/WorkflowStub.php +++ b/src/V2/WorkflowStub.php @@ -2132,6 +2132,17 @@ private function recordCancellationRequest( } $member = $scopePreparation === null ? null : collect($scopePreparation->childMembers) ->firstWhere('child_call_id', $scopeTarget['child_call_id']); + if ((\Workflow\V2\Support\ImportedLineageReferences::imported($scopeParent) + || \Workflow\V2\Support\ImportedLineageReferences::imported($run)) + && ($member === null || ! \Workflow\V2\Support\ImportedLineageReferences::allowsChild( + $scopeParent, + $run, + $member['sequence'], + $scopeTarget['child_call_id'], + true + ))) { + throw new LogicException('cancellation_scope_child_target_mismatch'); + } if ($scopePreparation === null || $scopePreparation->historyEventId !== $scopeTarget['preparation_history_event_id'] || $member === null || $member['child_workflow_instance_id'] !== $instance->id || $member['child_workflow_run_id'] !== $run->id || $member['cancellation_policy'] === 'abandon' @@ -5187,7 +5198,7 @@ private function createParentResumeTasks(WorkflowRun $childRun): array ->lockForUpdate() ->get(); - /** @var array $parentReferences */ + /** @var array $parentReferences */ $parentReferences = []; foreach ($parentLinks as $parentLink) { @@ -5196,6 +5207,7 @@ private function createParentResumeTasks(WorkflowRun $childRun): array } $parentReferences[$parentLink->parent_workflow_run_id] = [ + 'parent_workflow_instance_id' => $parentLink->parent_workflow_instance_id, 'parent_workflow_run_id' => $parentLink->parent_workflow_run_id, 'parent_sequence' => is_int($parentLink->sequence) ? $parentLink->sequence @@ -5203,7 +5215,7 @@ private function createParentResumeTasks(WorkflowRun $childRun): array ]; } - if ($parentReferences === []) { + if ($parentReferences === [] && ! \Workflow\V2\Support\ImportedLineageReferences::imported($childRun)) { $parentReference = ChildRunHistory::parentReferenceForRun($childRun); if ($parentReference !== null) { @@ -5214,15 +5226,24 @@ private function createParentResumeTasks(WorkflowRun $childRun): array foreach ($parentReferences as $parentReference) { /** @var WorkflowRun|null $parentRun */ $parentRun = self::runQuery() + ->where('namespace', $childRun->namespace) ->lockForUpdate() ->find($parentReference['parent_workflow_run_id']); - if ($parentRun === null || in_array($parentRun->status, [ - RunStatus::Completed, - RunStatus::Failed, - RunStatus::Cancelled, - RunStatus::Terminated, - ], true)) { + if ($parentRun === null || $parentRun->workflow_instance_id !== $parentReference['parent_workflow_instance_id'] + || ! \Workflow\V2\Support\ImportedLineageReferences::allowsChild( + $parentRun, + $childRun, + $parentReference['parent_sequence'], + $parentReference['child_call_id'] ?? null, + true + ) + || in_array($parentRun->status, [ + RunStatus::Completed, + RunStatus::Failed, + RunStatus::Cancelled, + RunStatus::Terminated, + ], true)) { continue; } diff --git a/tests/Feature/V2/V2ScopedChildCancellationTest.php b/tests/Feature/V2/V2ScopedChildCancellationTest.php index 4623a23fc..133663577 100644 --- a/tests/Feature/V2/V2ScopedChildCancellationTest.php +++ b/tests/Feature/V2/V2ScopedChildCancellationTest.php @@ -5,7 +5,9 @@ namespace Tests\Feature\V2; use Illuminate\Support\Carbon; +use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; +use Illuminate\Support\Str; use LogicException; use PHPUnit\Framework\Attributes\DataProvider; use Tests\Fixtures\V2\TestSignalWorkflow; @@ -26,6 +28,8 @@ use Workflow\V2\Support\CancellationScopeRequests; use Workflow\V2\Support\CooperativeCancellationDelivery; use Workflow\V2\Support\DefaultWorkflowTaskBridge; +use Workflow\V2\Support\EmbeddedV2HistoryImport; +use Workflow\V2\Support\HistoryExport; use Workflow\V2\Support\HistoryTimeline; use Workflow\V2\Support\ScopedChildCancellation; use Workflow\V2\Support\ScopedChildCancellationDelivery; @@ -51,6 +55,148 @@ protected function tearDown(): void parent::tearDown(); } + #[DataProvider('importedCounterparts')] + public function testImportedPreparationCannotCaptureAnUnrelatedLaterChild(bool $counterpart): void + { + [$run, $task, $scope, $member, $target] = $this->importedPair($counterpart); + $before = $target->fresh() + ->getAttributes(); + $refused = false; + try { + $request = CancellationScopeRequests::request($run, $scope, '1.20', 30); + $prepared = app(DefaultWorkflowTaskBridge::class)->prepareCancellationScopeDelivery( + $task->id, + $task->lease_owner, + $task->attempt_count, + $scope, + $request->payload['request_id'], + 4, + 'child', + protocolVersion: '1.20' + ); + if (! $prepared['prepared']) { + $this->assertSame('cancellation_scope_child_target_mismatch', $prepared['reason']); + $refused = true; + } else { + ScopedChildCancellationDelivery::request( + $run, + $task, + $member['child_call_id'], + $scope, + $prepared['request_id'], + '1.20' + ); + } + } catch (LogicException $error) { + $this->assertSame('cancellation_scope_child_target_mismatch', $error->getMessage()); + $refused = true; + } + if ($counterpart) { + $this->assertFalse($refused); + $this->assertNotNull($target->fresh()->cancellation_request_command_id); + } else { + $this->assertTrue($refused, 'An unrelated later run cannot become a scoped cancellation target.'); + $this->assertSame($before, $target->fresh()->getAttributes()); + $this->assertSame(0, $target->commands()->count()); + } + } + + public static function importedCounterparts(): iterable + { + yield 'unrelated later child' => [false]; + yield 'counterpart import' => [true]; + } + + #[DataProvider('importedEdgeChanges')] + public function testImportedScopedRequestRechecksItsExactEdge(string $change, string $actor): void + { + [$run, $task, $scope, $member, $target] = $this->importedPair(true); + $prepared = $this->prepare($run, $task, $scope); + $link = WorkflowLink::query()->where('parent_workflow_run_id', $run->id)->where( + 'child_workflow_run_id', + $target->id + )->sole(); + if ($change === 'missing') { + $link->delete(); + } elseif ($change === 'reversed') { + $link->forceFill([ + 'parent_workflow_run_id' => $target->id, + 'parent_workflow_instance_id' => $target->workflow_instance_id, + 'child_workflow_run_id' => $run->id, + 'child_workflow_instance_id' => $run->workflow_instance_id, + ])->save(); + } elseif ($change !== 'owned') { + $link->forceFill(match ($change) { + 'type' => [ + 'link_type' => 'continue_as_new', + ], + 'sequence' => [ + 'sequence' => 99, + ], + 'call' => [ + 'id' => (string) Str::ulid(), + ], + 'parent-instance' => [ + 'parent_workflow_instance_id' => $target->workflow_instance_id, + ], + 'child-instance' => [ + 'child_workflow_instance_id' => $run->workflow_instance_id, + ], + })->save(); + } + $before = $target->fresh() + ->getAttributes(); + $refused = false; + try { + if ($actor === 'delivery') { + ScopedChildCancellationDelivery::request( + $run, + $task, + $member['child_call_id'], + $scope, + $prepared['request_id'], + '1.20' + ); + } else { + WorkflowStub::loadRun($target->id)->attemptRequestCancellationFromScope( + $run->id, + $scope, + $prepared['preparation_history_event_id'], + $member['child_call_id'] + ); + } + } catch (LogicException $error) { + $this->assertSame('cancellation_scope_child_target_mismatch', $error->getMessage()); + $refused = true; + } + if ($change === 'owned') { + $this->assertFalse($refused); + $this->assertNotNull($target->fresh()->cancellation_request_command_id); + } else { + $this->assertTrue($refused, 'Imported scope effects must revalidate the directed call edge.'); + $this->assertSame($before, $target->fresh()->getAttributes()); + $this->assertSame(0, $target->commands()->count()); + } + } + + public static function importedEdgeChanges(): iterable + { + foreach ([ + 'missing', + 'reversed', + 'type', + 'sequence', + 'call', + 'parent-instance', + 'child-instance', + 'owned', + ] as $change) { + foreach (['delivery', 'direct'] as $actor) { + yield $change . '/' . $actor => [$change, $actor]; + } + } + } + public function testPreparationFreezesAllPoliciesWithoutTouchingChildrenSiblingsOrShields(): void { [$run, $task, $scope, $sibling, $shield] = $this->tree(); @@ -816,6 +962,88 @@ public function testChildInheritsTheEarlierPreparedParentRunDeadline(): void ScopedChildCancellation::assertReady(CancellationScopeDelivery::prepared($run->fresh(), $scope), $run->fresh()); } + /** + * @return array{WorkflowRun, WorkflowTask, string, array, WorkflowRun} + */ + private function importedPair(bool $counterpart): array + { + [$run, $task, $scope] = $this->tree(); + $member = $this->child($run, $task, $scope, 4); + $sourceChild = WorkflowRun::query()->findOrFail($member['child_workflow_run_id']); + $task->forceFill([ + 'status' => TaskStatus::Ready, + 'leased_at' => null, + 'lease_owner' => null, + 'lease_expires_at' => null, + ])->save(); + $parentBundle = HistoryExport::forRun($run->fresh()); + $childBundle = HistoryExport::forRun($sourceChild->fresh()); + foreach ([ + 'workflow_run_summaries', + 'workflow_run_waits', + 'workflow_run_timeline_entries', + 'workflow_run_timer_entries', + 'workflow_run_lineage_entries', + 'workflow_search_attributes', + 'workflow_memos', + 'workflow_history_events', + 'workflow_tasks', + 'activity_attempts', + 'activity_executions', + 'workflow_run_timers', + 'workflow_failures', + 'workflow_child_calls', + 'workflow_links', + 'workflow_signal_records', + 'workflow_updates', + 'workflow_commands', + 'workflow_runs', + 'workflow_instances', + ] as $table) { + DB::table($table)->delete(); + } + $this->assertSame('imported', EmbeddedV2HistoryImport::import($parentBundle, [ + 'namespace' => 'imported-scope', + ])['status']); + if ($counterpart) { + $this->assertSame('imported', EmbeddedV2HistoryImport::import($childBundle, [ + 'namespace' => 'imported-scope', + ])['status']); + } else { + WorkflowInstance::query()->create([ + 'id' => $member['child_workflow_instance_id'], + 'namespace' => 'imported-scope', + 'workflow_type' => 'unrelated-later', + 'workflow_class' => TestSignalWorkflow::class, + 'current_run_id' => $member['child_workflow_run_id'], + 'run_count' => 1, + ]); + $late = WorkflowRun::query()->create([ + 'id' => $member['child_workflow_run_id'], + 'workflow_instance_id' => $member['child_workflow_instance_id'], + 'namespace' => 'imported-scope', + 'workflow_type' => 'unrelated-later', + 'workflow_class' => TestSignalWorkflow::class, + 'status' => 'waiting', + 'run_number' => 1, + 'payload_codec' => 'avro', + 'arguments' => Serializer::serialize([]), + ]); + WorkflowHistoryEvent::record($late, HistoryEventType::WorkflowStarted, []); + } + $parent = WorkflowRun::query()->findOrFail($run->id); + $claim = WorkflowTask::query()->findOrFail($task->id); + $claim->forceFill([ + 'status' => TaskStatus::Leased, + 'lease_owner' => 'imported-scope-worker', + 'attempt_count' => 2, + 'lease_expires_at' => now() + ->addMinutes(5), + ])->save(); + + return [$parent, $claim, $scope, $member, WorkflowRun::query()->findOrFail($member['child_workflow_run_id'])]; + } + /** * @return array{WorkflowRun, WorkflowTask, string, string, string} */ diff --git a/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php b/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php index 4c53d2b0c..a7ba3a8e2 100644 --- a/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php +++ b/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php @@ -7,6 +7,7 @@ use Illuminate\Database\Events\TransactionBeginning; use Illuminate\Database\Events\TransactionCommitting; use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Queue; use Illuminate\Support\Str; use PHPUnit\Framework\Attributes\DataProvider; use RuntimeException; @@ -15,16 +16,23 @@ use Workflow\Serializers\CodecRegistry; use Workflow\Serializers\Serializer; use Workflow\V2\Enums\HistoryEventType; +use Workflow\V2\Enums\TaskStatus; +use Workflow\V2\Enums\TaskType; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; use Workflow\V2\Models\WorkflowLink; use Workflow\V2\Models\WorkflowRun; +use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Support\ChildRunHistory; use Workflow\V2\Support\EmbeddedV2HistoryImport; +use Workflow\V2\Support\EmbeddedV2ImportContract; use Workflow\V2\Support\HistoryExport; +use Workflow\V2\Support\ImportedLineageReferences; use Workflow\V2\Support\RunDetailView; use Workflow\V2\Support\RunLineageProjector; use Workflow\V2\Support\RunLineageView; +use Workflow\V2\Support\WorkflowExecutor; +use Workflow\V2\WorkflowStub; final class EmbeddedV2HistoryImportLineageTest extends TestCase { @@ -323,6 +331,268 @@ public static function lateEndpoints(): iterable } } + #[DataProvider('selfLineageReferences')] + public function testSelfLineageIsRejectedWithoutChangingTheStore(string $reference, bool $dryRun): void + { + $bundle = $this->sourcePair()['parent']; + $ownRun = $bundle['workflow']['run_id']; + $ownInstance = $bundle['workflow']['instance_id']; + if (in_array($reference, ['parents', 'children'], true)) { + $link = $bundle['links']['children'][0]; + foreach (['parent', 'child'] as $direction) { + $link[$direction . '_workflow_run_id'] = $ownRun; + $link[$direction . '_workflow_instance_id'] = $ownInstance; + } + $bundle['links'] = [ + 'parents' => [], + 'children' => [], + ]; + $bundle['links'][$reference] = [$link]; + } elseif ($reference === 'started-parent') { + $bundle['history_events'][0]['payload']['parent_workflow_run_id'] = $ownRun; + $bundle['history_events'][0]['payload']['parent_workflow_instance_id'] = $ownInstance; + } elseif ($reference === 'continued-from') { + $bundle['history_events'][0]['payload']['continued_from_run_id'] = $ownRun; + } elseif (in_array($reference, ['scheduled-child', 'started-child'], true)) { + $index = $reference === 'scheduled-child' ? 1 : 2; + $bundle['history_events'][$index]['payload']['child_workflow_run_id'] = $ownRun; + $bundle['history_events'][$index]['payload']['child_workflow_instance_id'] = $ownInstance; + } else { + $event = $bundle['history_events'][2]; + $event['id'] = (string) Str::ulid(); + $event['sequence']++; + $event['type'] = $reference; + $event['payload'] = $reference === 'WorkflowContinuedAsNew' + ? [ + 'continued_to_run_id' => $ownRun, + ] + : [ + ...$event['payload'], + 'child_workflow_run_id' => $ownRun, + 'child_workflow_instance_id' => $ownInstance, + ]; + $bundle['history_events'][] = $event; + $bundle['workflow']['last_history_sequence'] = $event['sequence']; + } + $bundle = $this->reseal($bundle); + $this->clearState(); + $this->createRun('unrelated-existing', (string) Str::ulid(), 'tenant-b', 'unrelated-existing'); + $before = $this->snapshot(); + + $report = EmbeddedV2HistoryImport::import($bundle, [ + 'namespace' => 'tenant-a', + 'dry_run' => $dryRun, + ]); + + $this->assertSame('rejected', $report['status']); + $this->assertContains( + 'references.lineage_self_reference', + array_column($report['eligibility']['errors'], 'rule') + ); + $this->assertSame(0, array_sum($report['rows'])); + $this->assertSame($before, $this->snapshot()); + } + + public static function selfLineageReferences(): iterable + { + foreach (['parents', 'children', 'started-parent', 'continued-from', 'WorkflowContinuedAsNew', + 'scheduled-child', 'started-child', 'ChildRunCompleted', 'ChildRunFailed', 'ChildRunCancelled', + 'ChildRunTerminated'] as $reference) { + foreach ([false, true] as $preview) { + yield $reference . ($preview ? '/preview' : '/write') => [$reference, $preview]; + } + } + } + + public function testImportedLineageResolverDoesNotAuthorizeItsOwnRun(): void + { + $run = new WorkflowRun([ + 'id' => (string) Str::ulid(), + 'workflow_instance_id' => 'self-instance', + 'namespace' => 'tenant-a', + 'import_source' => EmbeddedV2ImportContract::IMPORT_SOURCE, + ]); + $this->assertFalse(ImportedLineageReferences::allows($run, $run)); + } + + #[DataProvider('parentResumePaths')] + public function testImportedChildClosureCannotResumeAnUnrelatedLaterParent( + string $path, + string $namespace, + bool $counterpart + ): void { + Queue::fake(); + config([ + 'queue.default' => 'database', + 'workflows.v2.task_dispatch_mode' => 'poll', + ]); + $bundles = $this->sourcePair(); + $this->clearState(); + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundles['child'], [ + 'namespace' => 'tenant-a', + ])['status']); + if ($counterpart) { + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundles['parent'], [ + 'namespace' => 'tenant-a', + ])['status']); + $parent = WorkflowRun::query()->findOrFail($bundles['parent']['workflow']['run_id']); + $observed = [$parent]; + } else { + $identity = $bundles['parent']['workflow']; + $parent = $this->createRun( + $identity['instance_id'], + $identity['run_id'], + $namespace, + $identity['workflow_type'] + ); + $otherChild = $this->createRun('unrelated-child', (string) Str::ulid(), $namespace, 'unrelated-child'); + $call = (string) Str::ulid(); + $payload = [ + 'sequence' => 1, + 'child_call_id' => $call, + 'workflow_link_id' => $call, + 'child_workflow_instance_id' => $otherChild->workflow_instance_id, + 'child_workflow_run_id' => $otherChild->id, + 'child_workflow_type' => 'unrelated-child', + ]; + WorkflowHistoryEvent::record($parent, HistoryEventType::WorkflowStarted, []); + WorkflowHistoryEvent::record($parent, HistoryEventType::ChildWorkflowScheduled, $payload); + WorkflowHistoryEvent::record($parent, HistoryEventType::ChildRunStarted, $payload); + WorkflowLink::query()->create([ + 'id' => $call, + 'link_type' => 'child_workflow', + 'sequence' => 1, + 'parent_workflow_instance_id' => $parent->workflow_instance_id, + 'parent_workflow_run_id' => $parent->id, + 'child_workflow_instance_id' => $otherChild->workflow_instance_id, + 'child_workflow_run_id' => $otherChild->id, + ]); + $observed = [$parent, $otherChild]; + } + $child = WorkflowRun::query()->findOrFail($bundles['child']['workflow']['run_id']); + $this->assertSame( + $counterpart ? 1 : 0, + WorkflowLink::query()->where('child_workflow_run_id', $child->id)->count() + ); + $before = $this->ownedState($observed); + + if ($path === 'operator') { + $this->assertTrue(WorkflowStub::loadRun($child->id)->attemptCancel('imported child closure')->accepted()); + } else { + $child->forceFill([ + 'run_deadline_at' => now() + ->subSecond(), + ])->save(); + $task = WorkflowTask::query()->create([ + 'workflow_run_id' => $child->id, + 'namespace' => $child->namespace, + 'task_type' => TaskType::Workflow, + 'status' => TaskStatus::Leased, + 'lease_owner' => 'imported-child-worker', + 'attempt_count' => 1, + 'lease_expires_at' => now() + ->addMinute(), + 'available_at' => now(), + 'payload' => [], + ]); + $this->assertTrue( + DB::transaction(static fn (): bool => (new WorkflowExecutor())->timeoutIfDeadlineExpired( + $child->fresh(), + $task + )) + ); + } + + $this->assertTrue($child->fresh()->status->isTerminal()); + if ($counterpart) { + $this->assertSame(1, $parent->historyEvents()->where('payload->child_workflow_run_id', $child->id) + ->whereIn( + 'event_type', + [HistoryEventType::ChildRunCancelled->value, HistoryEventType::ChildRunFailed->value] + )->count()); + $this->assertSame(1, $parent->tasks()->where('task_type', TaskType::Workflow)->count()); + } else { + $this->assertSame($before, $this->ownedState($observed)); + } + } + + public static function parentResumePaths(): iterable + { + foreach (['operator', 'worker'] as $path) { + foreach (['tenant-a', 'tenant-b'] as $namespace) { + yield $path . '/' . $namespace . '/unrelated-later' => [$path, $namespace, false]; + } + yield $path . '/counterpart-import' => [$path, 'tenant-a', true]; + } + } + + /** @param list $runs + * @return array + */ + private function ownedState(array $runs): array + { + $ids = array_map(static fn (WorkflowRun $run): string => $run->id, $runs); + $instances = array_map(static fn (WorkflowRun $run): string => $run->workflow_instance_id, $runs); + $state = []; + foreach (self::TABLES as $table) { + $query = DB::table($table); + if ($table === 'workflow_instances') { + $query->whereIn('id', $instances); + } elseif ($table === 'workflow_runs' || $table === 'workflow_run_summaries') { + $query->whereIn('id', $ids); + } elseif ($table === 'workflow_links') { + $query->where( + static fn ($builder) => $builder->whereIn( + 'parent_workflow_run_id', + $ids + )->orWhereIn('child_workflow_run_id', $ids) + ); + } else { + $query->whereIn('workflow_run_id', $ids); + } + $state[$table] = $query->get()->map(static fn (object $row): array => (array) $row)->all(); + } + + return $state; + } + + /** @param array $bundle + * @return array + */ + private function reseal(array $bundle): array + { + unset($bundle['integrity']); + $canonical = static function (mixed $value) use (&$canonical): mixed { + if (! is_array($value)) { + return $value; + } + foreach ($value as $key => $entry) { + $value[$key] = $canonical($entry); + } + if (! array_is_list($value)) { + ksort($value, SORT_STRING); + } + + return $value; + }; + $bundle['integrity'] = [ + 'canonicalization' => 'json-recursive-ksort-v1', + 'checksum_algorithm' => 'sha256', + 'checksum' => hash( + 'sha256', + json_encode( + $canonical($bundle), + JSON_UNESCAPED_SLASHES | JSON_PRESERVE_ZERO_FRACTION | JSON_THROW_ON_ERROR + ) + ), + 'signature_algorithm' => null, + 'signature' => null, + 'key_id' => null, + ]; + + return $bundle; + } + /** * @return array{parent: array, child: array} */ From a6f9195ba74a7dd601943efb3467c8bf6696427e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 26 Oct 2026 00:59:52 +0000 Subject: [PATCH 6/6] Fence imported lineage across continuation --- src/V2/Support/DefaultWorkflowTaskBridge.php | 47 +++-- src/V2/Support/WorkflowExecutor.php | 44 +++- .../V2/EmbeddedV2HistoryImportLineageTest.php | 195 +++++++++++++++++- 3 files changed, 261 insertions(+), 25 deletions(-) diff --git a/src/V2/Support/DefaultWorkflowTaskBridge.php b/src/V2/Support/DefaultWorkflowTaskBridge.php index 69e3e6594..b5df981b1 100644 --- a/src/V2/Support/DefaultWorkflowTaskBridge.php +++ b/src/V2/Support/DefaultWorkflowTaskBridge.php @@ -5048,8 +5048,30 @@ private function applyContinueAsNew( ->get(); $parentRunsToProject = []; + $trustedParentReference = null; foreach ($parentChildLinks as $parentChildLink) { + /** @var WorkflowRun|null $parentRun */ + $parentRun = is_string($parentChildLink->parent_workflow_run_id) + && $parentChildLink->parent_workflow_run_id !== '' + ? ConfiguredV2Models::query('run_model', WorkflowRun::class) + ->lockForUpdate() + ->find($parentChildLink->parent_workflow_run_id) + : null; + $requiresImportedEdge = ImportedLineageReferences::imported($run) + || ($parentRun instanceof WorkflowRun && ImportedLineageReferences::imported($parentRun)); + if ($requiresImportedEdge && (! $parentRun instanceof WorkflowRun + || ! is_int($parentChildLink->sequence) + || ! ImportedLineageReferences::allowsChild( + $parentRun, + $run, + $parentChildLink->sequence, + $childCallId, + true, + ))) { + continue; + } + /** @var WorkflowLink $continuedChildLink */ $continuedChildLink = WorkflowLink::query()->create([ 'link_type' => 'child_workflow', @@ -5063,20 +5085,17 @@ private function applyContinueAsNew( 'parent_close_policy' => $parentChildLink->parent_close_policy, ]); - if ( - ! is_string($parentChildLink->parent_workflow_run_id) - || $parentChildLink->parent_workflow_run_id === '' - || ! is_int($parentChildLink->sequence) - ) { - continue; + if (ImportedLineageReferences::imported($run) && $parentRun instanceof WorkflowRun + && is_int($parentChildLink->sequence) && $trustedParentReference === null) { + $trustedParentReference = [ + 'parent_workflow_instance_id' => $parentRun->workflow_instance_id, + 'parent_workflow_run_id' => $parentRun->id, + 'parent_sequence' => $parentChildLink->sequence, + ]; } - /** @var WorkflowRun|null $parentRun */ - $parentRun = ConfiguredV2Models::query('run_model', WorkflowRun::class) - ->lockForUpdate() - ->find($parentChildLink->parent_workflow_run_id); - - if ($parentRun === null || $parentRun->status->isTerminal()) { + if (! $parentRun instanceof WorkflowRun || ! is_int($parentChildLink->sequence) + || $parentRun->status->isTerminal()) { continue; } @@ -5118,7 +5137,9 @@ private function applyContinueAsNew( ], $parallelMetadata), static fn (mixed $value): bool => $value !== null)); } - $parentReference = ChildRunHistory::parentReferenceForRun($run); + $parentReference = ImportedLineageReferences::imported($run) + ? $trustedParentReference + : ChildRunHistory::parentReferenceForRun($run); WorkflowHistoryEvent::record($continuedRun, HistoryEventType::WorkflowStarted, [ 'workflow_class' => $workflowType, diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index f2b2faeab..5172d530e 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -3763,8 +3763,30 @@ private function continueAsNew( ->get(); $parentRunsToProject = []; + $trustedParentReference = null; foreach ($parentChildLinks as $parentChildLink) { + /** @var WorkflowRun|null $parentRun */ + $parentRun = is_string($parentChildLink->parent_workflow_run_id) + && $parentChildLink->parent_workflow_run_id !== '' + ? WorkflowRun::query() + ->lockForUpdate() + ->find($parentChildLink->parent_workflow_run_id) + : null; + $requiresImportedEdge = ImportedLineageReferences::imported($run) + || ($parentRun instanceof WorkflowRun && ImportedLineageReferences::imported($parentRun)); + if ($requiresImportedEdge && (! $parentRun instanceof WorkflowRun + || ! is_int($parentChildLink->sequence) + || ! ImportedLineageReferences::allowsChild( + $parentRun, + $run, + $parentChildLink->sequence, + $childCallId, + true, + ))) { + continue; + } + $continuedChildLink = WorkflowLink::query()->create([ 'link_type' => 'child_workflow', 'sequence' => $parentChildLink->sequence, @@ -3777,20 +3799,18 @@ private function continueAsNew( 'parent_close_policy' => $parentChildLink->parent_close_policy, ]); - if ( - ! is_string($parentChildLink->parent_workflow_run_id) - || $parentChildLink->parent_workflow_run_id === '' - ) { - continue; + if (ImportedLineageReferences::imported($run) && $parentRun instanceof WorkflowRun + && is_int($parentChildLink->sequence) && $trustedParentReference === null) { + $trustedParentReference = [ + 'parent_workflow_instance_id' => $parentRun->workflow_instance_id, + 'parent_workflow_run_id' => $parentRun->id, + 'parent_sequence' => $parentChildLink->sequence, + ]; } - /** @var WorkflowRun|null $parentRun */ - $parentRun = WorkflowRun::query() - ->lockForUpdate() - ->find($parentChildLink->parent_workflow_run_id); - if ( ! $parentRun instanceof WorkflowRun + || ! is_int($parentChildLink->sequence) || in_array($parentRun->status, [ RunStatus::Completed, RunStatus::Failed, @@ -3840,7 +3860,9 @@ private function continueAsNew( ], $parallelMetadata), static fn ($value): bool => $value !== null)); } - $parentReference = ChildRunHistory::parentReferenceForRun($run); + $parentReference = ImportedLineageReferences::imported($run) + ? $trustedParentReference + : ChildRunHistory::parentReferenceForRun($run); $continuedMemo = AvroValueJsonProjection::project($continuedRun->typedMemos()); $continuedSearchAttributes = $continuedRun->typedSearchAttributes(); diff --git a/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php b/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php index a7ba3a8e2..00184b427 100644 --- a/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php +++ b/tests/Unit/V2/EmbeddedV2HistoryImportLineageTest.php @@ -15,7 +15,11 @@ use Throwable; use Workflow\Serializers\CodecRegistry; use Workflow\Serializers\Serializer; +use Workflow\V2\Attributes\Type; +use function Workflow\V2\continueAsNew; +use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; +use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; use Workflow\V2\Models\WorkflowHistoryEvent; @@ -32,6 +36,8 @@ use Workflow\V2\Support\RunLineageProjector; use Workflow\V2\Support\RunLineageView; use Workflow\V2\Support\WorkflowExecutor; +use Workflow\V2\Workflow; + use Workflow\V2\WorkflowStub; final class EmbeddedV2HistoryImportLineageTest extends TestCase @@ -397,7 +403,7 @@ public static function selfLineageReferences(): iterable { foreach (['parents', 'children', 'started-parent', 'continued-from', 'WorkflowContinuedAsNew', 'scheduled-child', 'started-child', 'ChildRunCompleted', 'ChildRunFailed', 'ChildRunCancelled', - 'ChildRunTerminated'] as $reference) { + 'ChildRunTerminated', 'ChildRunTimedOut'] as $reference) { foreach ([false, true] as $preview) { yield $reference . ($preview ? '/preview' : '/write') => [$reference, $preview]; } @@ -526,6 +532,184 @@ public static function parentResumePaths(): iterable } } + #[DataProvider('importedContinuationPaths')] + public function testImportedChildContinuationRetainsOnlyAnOwnedParentEdge( + string $path, + string $parentNamespace, + bool $counterpart, + ): void { + Queue::fake(); + config([ + 'queue.default' => 'database', + 'workflows.v2.task_dispatch_mode' => 'poll', + ]); + $bundles = $this->sourcePair(); + $this->clearState(); + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundles['child'], [ + 'namespace' => 'tenant-a', + ])['status']); + if ($counterpart) { + $this->assertSame('imported', EmbeddedV2HistoryImport::import($bundles['parent'], [ + 'namespace' => 'tenant-a', + ])['status']); + $parent = WorkflowRun::query()->findOrFail($bundles['parent']['workflow']['run_id']); + $observed = [$parent]; + } else { + $identity = $bundles['parent']['workflow']; + $parent = $this->createRun( + $identity['instance_id'], + $identity['run_id'], + $parentNamespace, + $identity['workflow_type'], + ); + $otherChild = $this->createRun( + 'unrelated-continuation-child', + (string) Str::ulid(), + $parentNamespace, + 'unrelated-continuation-child', + ); + $call = (string) Str::ulid(); + $payload = [ + 'sequence' => 1, + 'child_call_id' => $call, + 'workflow_link_id' => $call, + 'child_workflow_instance_id' => $otherChild->workflow_instance_id, + 'child_workflow_run_id' => $otherChild->id, + 'child_workflow_type' => 'unrelated-continuation-child', + ]; + WorkflowHistoryEvent::record($parent, HistoryEventType::WorkflowStarted, []); + WorkflowHistoryEvent::record($parent, HistoryEventType::ChildWorkflowScheduled, $payload); + WorkflowHistoryEvent::record($parent, HistoryEventType::ChildRunStarted, $payload); + WorkflowLink::query()->create([ + 'id' => $call, + 'link_type' => 'child_workflow', + 'sequence' => 1, + 'parent_workflow_instance_id' => $parent->workflow_instance_id, + 'parent_workflow_run_id' => $parent->id, + 'child_workflow_instance_id' => $otherChild->workflow_instance_id, + 'child_workflow_run_id' => $otherChild->id, + ]); + $observed = [$parent, $otherChild]; + } + $child = WorkflowRun::query()->findOrFail($bundles['child']['workflow']['run_id']); + $child->forceFill([ + 'workflow_class' => ImportedLineageContinuationWorkflow::class, + 'workflow_type' => 'imported-lineage-continuation-workflow', + 'arguments' => Serializer::serializeWithCodec($child->payload_codec, [true]), + ])->save(); + $child->instance->forceFill([ + 'workflow_class' => ImportedLineageContinuationWorkflow::class, + 'workflow_type' => 'imported-lineage-continuation-workflow', + ])->save(); + $before = $this->ownedState($observed); + + $continued = $this->continueImportedChild($child, $path); + $this->closeContinuedRun($continued); + + $this->assertSame(RunStatus::Completed, $continued->fresh()->status); + if (! $counterpart) { + $this->assertSame($before, $this->ownedState($observed)); + $started = $continued->historyEvents() + ->where('event_type', HistoryEventType::WorkflowStarted->value) + ->oldest('sequence') + ->first(); + $this->assertNull($started?->payload['parent_workflow_run_id'] ?? null); + $this->assertSame(0, WorkflowLink::query() + ->where('parent_workflow_run_id', $parent->id) + ->where('child_workflow_run_id', $continued->id) + ->count()); + + return; + } + $this->assertSame(1, WorkflowLink::query() + ->where('parent_workflow_run_id', $parent->id) + ->where('child_workflow_run_id', $continued->id) + ->where('link_type', 'child_workflow') + ->count()); + $started = $continued->historyEvents() + ->where('event_type', HistoryEventType::WorkflowStarted->value) + ->oldest('sequence') + ->first(); + $this->assertSame($parent->id, $started?->payload['parent_workflow_run_id'] ?? null); + $this->assertSame(1, $parent->historyEvents() + ->where('event_type', HistoryEventType::ChildRunCompleted->value) + ->where('payload->child_workflow_run_id', $continued->id) + ->count()); + $this->assertSame(1, $parent->tasks()->where('task_type', TaskType::Workflow->value)->count()); + } + + /** + * @return iterable + */ + public static function importedContinuationPaths(): iterable + { + foreach (['bridge', 'executor'] as $path) { + yield $path . '/same-namespace-unrelated' => [$path, 'tenant-a', false]; + yield $path . '/cross-namespace-unrelated' => [$path, 'tenant-b', false]; + yield $path . '/counterpart-import' => [$path, 'tenant-a', true]; + } + } + + private function continueImportedChild(WorkflowRun $child, string $path): WorkflowRun + { + $task = $this->leasedTask($child, 'imported-continuation-' . $path); + if ($path === 'bridge') { + $result = $this->app->make(WorkflowTaskBridge::class)->complete($task->id, [[ + 'type' => 'continue_as_new', + 'workflow_type' => ImportedLineageContinuationWorkflow::class, + 'arguments' => Serializer::serializeWithCodec($child->payload_codec, [false]), + 'payload_codec' => $child->payload_codec, + ]]); + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + } else { + DB::transaction(static fn (): ?WorkflowTask => (new WorkflowExecutor())->run( + $child->fresh(), + $task->fresh(), + )); + } + + return WorkflowRun::query() + ->where('workflow_instance_id', $child->workflow_instance_id) + ->whereKeyNot($child->id) + ->sole(); + } + + private function closeContinuedRun(WorkflowRun $continued): void + { + $task = $continued->tasks() + ->where('task_type', TaskType::Workflow->value)->sole(); + $task->forceFill([ + 'status' => TaskStatus::Leased->value, + 'lease_owner' => 'continued-run-closer', + 'attempt_count' => 1, + 'lease_expires_at' => now() + ->addMinute(), + ])->save(); + DB::transaction(static fn (): ?WorkflowTask => (new WorkflowExecutor())->run( + $continued->fresh(), + $task->fresh(), + )); + } + + private function leasedTask(WorkflowRun $run, string $owner): WorkflowTask + { + return WorkflowTask::query()->create([ + 'workflow_run_id' => $run->id, + 'namespace' => $run->namespace, + 'task_type' => TaskType::Workflow->value, + 'status' => TaskStatus::Leased->value, + 'lease_owner' => $owner, + 'attempt_count' => 1, + 'lease_expires_at' => now() + ->addMinute(), + 'available_at' => now(), + 'payload' => [], + 'connection' => $run->connection, + 'queue' => $run->queue, + 'compatibility' => $run->compatibility, + ]); + } + /** @param list $runs * @return array */ @@ -695,3 +879,12 @@ private function snapshot(): array return $result; } } + +#[Type('imported-lineage-continuation-workflow')] +final class ImportedLineageContinuationWorkflow extends Workflow +{ + public function handle(bool $continue): mixed + { + return $continue ? continueAsNew(false) : 'continued child completed'; + } +}