From e0fb9505d48e7e96c2e601e79dd9d52b92a919a3 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 02:20:11 +0000 Subject: [PATCH 1/4] Draft the entity reference adapter/resolver API with failing tests (#996) Experimental API under Workflow\V2\EntityReferences, following the decisions on #996: EntityReferenceAdapter (supports/identify/version), EntityReferenceResolver (resolve with a trusted EntityReferenceContext), the portable EntityReference {type, id, version}, EntityNotFound / EntityAccessDenied for resolvers, and the non-retryable EntityReferenceRefused (unknown type, unsupported version, parameter mismatch). Registration is workflows.v2.entity_references by logical type. V2EntityReferenceTest pins the accepted behavior with a plain Order entity: the activity receives the current persisted entity, history records {type, id, version} plus entity_reference_positions, replay never resolves, a missing entity fails with EntityNotFound, an unknown recorded type is refused non-retryably before the activity runs, and an unregistered object still fails. The first four fail until the implementation lands; this is a draft. Co-Authored-By: Claude Opus 5.5 --- .../EntityReferences/EntityAccessDenied.php | 22 +++ src/V2/EntityReferences/EntityNotFound.php | 22 +++ src/V2/EntityReferences/EntityReference.php | 60 +++++++ .../EntityReferenceAdapter.php | 41 +++++ .../EntityReferenceContext.php | 29 +++ .../EntityReferenceRefused.php | 30 ++++ .../EntityReferenceResolver.php | 28 +++ src/config/workflows.php | 11 ++ tests/Feature/V2/V2EntityReferenceTest.php | 165 ++++++++++++++++++ tests/Fixtures/V2/EntityReferences/Order.php | 17 ++ .../OrderReferenceAdapter.php | 26 +++ .../OrderReferenceResolver.php | 18 ++ .../V2/EntityReferences/OrderRepository.php | 37 ++++ .../EntityReferences/OrderStatusActivity.php | 17 ++ .../EntityReferences/OrderStatusWorkflow.php | 19 ++ 15 files changed, 542 insertions(+) create mode 100644 src/V2/EntityReferences/EntityAccessDenied.php create mode 100644 src/V2/EntityReferences/EntityNotFound.php create mode 100644 src/V2/EntityReferences/EntityReference.php create mode 100644 src/V2/EntityReferences/EntityReferenceAdapter.php create mode 100644 src/V2/EntityReferences/EntityReferenceContext.php create mode 100644 src/V2/EntityReferences/EntityReferenceRefused.php create mode 100644 src/V2/EntityReferences/EntityReferenceResolver.php create mode 100644 tests/Feature/V2/V2EntityReferenceTest.php create mode 100644 tests/Fixtures/V2/EntityReferences/Order.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderReferenceAdapter.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderReferenceResolver.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderRepository.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderStatusActivity.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php diff --git a/src/V2/EntityReferences/EntityAccessDenied.php b/src/V2/EntityReferences/EntityAccessDenied.php new file mode 100644 index 00000000..c440a3d6 --- /dev/null +++ b/src/V2/EntityReferences/EntityAccessDenied.php @@ -0,0 +1,22 @@ +type, $reference->id)); + } +} diff --git a/src/V2/EntityReferences/EntityNotFound.php b/src/V2/EntityReferences/EntityNotFound.php new file mode 100644 index 00000000..ce2120c0 --- /dev/null +++ b/src/V2/EntityReferences/EntityNotFound.php @@ -0,0 +1,22 @@ +type, $reference->id)); + } +} diff --git a/src/V2/EntityReferences/EntityReference.php b/src/V2/EntityReferences/EntityReference.php new file mode 100644 index 00000000..fd75fc5c --- /dev/null +++ b/src/V2/EntityReferences/EntityReference.php @@ -0,0 +1,60 @@ += 1.'); + } + } + + /** + * @return array{type: string, id: string, version: int} + */ + public function toArray(): array + { + return [ + 'type' => $this->type, + 'id' => $this->id, + 'version' => $this->version, + ]; + } + + /** + * @param array $value + */ + public static function fromArray(array $value): self + { + if (array_keys($value) !== ['type', 'id', 'version'] + || ! is_string($value['type']) || ! is_string($value['id']) || ! is_int($value['version'])) { + throw new InvalidArgumentException('A recorded entity reference must be exactly {type, id, version}.'); + } + + return new self($value['type'], $value['id'], $value['version']); + } +} diff --git a/src/V2/EntityReferences/EntityReferenceAdapter.php b/src/V2/EntityReferences/EntityReferenceAdapter.php new file mode 100644 index 00000000..938bf095 --- /dev/null +++ b/src/V2/EntityReferences/EntityReferenceAdapter.php @@ -0,0 +1,41 @@ + App\Exceptions\InvoiceDeclined::class, ], ], + // Experimental (workflow issue 996): pass a domain entity to an activity + // as a portable reference. Each logical type names an adapter that + // turns the entity into {type, id, version} when the activity is + // scheduled, and a resolver that loads the current entity before each + // attempt. Top-level activity arguments of in-process workers only. + 'entity_references' => [ + // 'orders.order' => [ + // 'adapter' => App\Infrastructure\OrderReferenceAdapter::class, + // 'resolver' => App\Infrastructure\OrderReferenceResolver::class, + // ], + ], // Worker-compatibility markers let you pin workflow runs to a specific // worker build and block incompatible workers from claiming tasks. All // three keys default to null ("no marker required"), which is the right diff --git a/tests/Feature/V2/V2EntityReferenceTest.php b/tests/Feature/V2/V2EntityReferenceTest.php new file mode 100644 index 00000000..82d4d1d3 --- /dev/null +++ b/tests/Feature/V2/V2EntityReferenceTest.php @@ -0,0 +1,165 @@ +set('queue.default', 'redis'); + config() + ->set('workflows.v2.entity_references', [ + 'orders.order' => [ + 'adapter' => OrderReferenceAdapter::class, + 'resolver' => OrderReferenceResolver::class, + ], + ]); + OrderRepository::reset(); + } + + public function testAnActivityReceivesTheCurrentEntityResolvedFromAPortableReference(): void + { + OrderRepository::save(new Order('o-1', 'paid')); + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-current'); + $workflow->start('o-1'); + $runId = $workflow->runId(); + + $this->runReadyTask($runId, TaskType::Workflow); + + // History holds the reference and its recorded position, not the entity. + $scheduled = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $runId) + ->where('event_type', HistoryEventType::ActivityScheduled->value) + ->sole(); + $execution = ActivityExecution::query()->where('workflow_run_id', $runId)->sole(); + $this->assertSame([[ + 'type' => 'orders.order', + 'id' => 'o-1', + 'version' => 1, + ], 'checked'], Serializer::unserialize($execution->arguments)); + $this->assertSame([0], $execution->activity_options['entity_reference_positions'] ?? null); + $this->assertSame([0], $scheduled->payload['activity']['entity_reference_positions'] ?? null); + + $this->runReadyTask($runId, TaskType::Activity); + $this->runReadyTask($runId, TaskType::Workflow); + + // The activity saw the persisted state, not the workflow's copy. + $this->assertSame('o-1:paid:checked', $workflow->refresh()->output()); + $this->assertSame(1, OrderRepository::$loads); + } + + public function testReplayNeverResolvesTheReference(): void + { + OrderRepository::save(new Order('o-2', 'paid')); + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-replay'); + $workflow->start('o-2'); + $runId = $workflow->runId(); + $this->runReadyTask($runId, TaskType::Workflow); + $this->runReadyTask($runId, TaskType::Activity); + $this->runReadyTask($runId, TaskType::Workflow); + $loads = OrderRepository::$loads; + + (new WorkflowReplayer())->replay(WorkflowRun::query()->findOrFail($runId)); + + $this->assertSame($loads, OrderRepository::$loads); + } + + public function testAMissingEntityFailsTheAttemptWithEntityNotFound(): void + { + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-missing'); + $workflow->start('o-missing'); + $runId = $workflow->runId(); + $this->runReadyTask($runId, TaskType::Workflow); + $this->runReadyTask($runId, TaskType::Activity); + + $failed = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $runId) + ->where('event_type', HistoryEventType::ActivityFailed->value) + ->sole(); + $this->assertSame(EntityNotFound::class, $failed->payload['exception_class'] ?? null); + } + + public function testAnUnknownRecordedTypeIsRefusedBeforeTheActivityRunsAndNotRetried(): void + { + OrderRepository::save(new Order('o-3', 'paid')); + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-unknown-type'); + $workflow->start('o-3'); + $runId = $workflow->runId(); + $this->runReadyTask($runId, TaskType::Workflow); + + // The worker that runs the attempt no longer knows the type. + config() + ->set('workflows.v2.entity_references', []); + $this->runReadyTask($runId, TaskType::Activity); + + $failed = WorkflowHistoryEvent::query() + ->where('workflow_run_id', $runId) + ->where('event_type', HistoryEventType::ActivityFailed->value) + ->sole(); + $this->assertSame(EntityReferenceRefused::class, $failed->payload['exception_class'] ?? null); + $this->assertTrue((bool) ($failed->payload['non_retryable'] ?? false)); + $this->assertSame(0, OrderRepository::$loads); + } + + public function testAnUnregisteredObjectStillFailsWithoutObjectSerialization(): void + { + config() + ->set('workflows.v2.entity_references', []); + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-unregistered'); + $workflow->start('o-4'); + $runId = $workflow->runId(); + $this->runReadyTask($runId, TaskType::Workflow); + + $this->assertSame(0, ActivityExecution::query()->where('workflow_run_id', $runId)->count()); + $this->assertTrue(WorkflowHistoryEvent::query() + ->where('workflow_run_id', $runId) + ->where('event_type', HistoryEventType::WorkflowFailed->value) + ->exists()); + } + + private function runReadyTask(string $runId, TaskType $taskType): void + { + $task = WorkflowTask::query() + ->where('workflow_run_id', $runId) + ->where('task_type', $taskType->value) + ->where('status', TaskStatus::Ready->value) + ->sole(); + + $job = $taskType === TaskType::Activity + ? new RunActivityTask($task->id) + : new RunWorkflowTask($task->id); + + $this->app->call([$job, 'handle']); + } +} diff --git a/tests/Fixtures/V2/EntityReferences/Order.php b/tests/Fixtures/V2/EntityReferences/Order.php new file mode 100644 index 00000000..239a16f5 --- /dev/null +++ b/tests/Fixtures/V2/EntityReferences/Order.php @@ -0,0 +1,17 @@ +id; + } + + public function version(): int + { + return 1; + } +} diff --git a/tests/Fixtures/V2/EntityReferences/OrderReferenceResolver.php b/tests/Fixtures/V2/EntityReferences/OrderReferenceResolver.php new file mode 100644 index 00000000..7d5ea149 --- /dev/null +++ b/tests/Fixtures/V2/EntityReferences/OrderReferenceResolver.php @@ -0,0 +1,18 @@ +id) ?? throw EntityNotFound::for($reference); + } +} diff --git a/tests/Fixtures/V2/EntityReferences/OrderRepository.php b/tests/Fixtures/V2/EntityReferences/OrderRepository.php new file mode 100644 index 00000000..dbb005f8 --- /dev/null +++ b/tests/Fixtures/V2/EntityReferences/OrderRepository.php @@ -0,0 +1,37 @@ + + */ + public static array $orders = []; + + public static int $loads = 0; + + public static function reset(): void + { + self::$orders = []; + self::$loads = 0; + } + + public static function save(Order $order): void + { + self::$orders[$order->id] = $order; + } + + public static function find(string $id): ?Order + { + ++self::$loads; + + return self::$orders[$id] ?? null; + } +} diff --git a/tests/Fixtures/V2/EntityReferences/OrderStatusActivity.php b/tests/Fixtures/V2/EntityReferences/OrderStatusActivity.php new file mode 100644 index 00000000..89cb266c --- /dev/null +++ b/tests/Fixtures/V2/EntityReferences/OrderStatusActivity.php @@ -0,0 +1,17 @@ +id}:{$order->status}:{$note}"; + } +} diff --git a/tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php b/tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php new file mode 100644 index 00000000..a1ad0cb6 --- /dev/null +++ b/tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php @@ -0,0 +1,19 @@ + Date: Fri, 13 Nov 2026 10:14:55 +0000 Subject: [PATCH 2/4] Treat entity reference maps independently of key order --- src/V2/EntityReferences/EntityReference.php | 5 +- tests/Unit/V2/EntityReferenceTest.php | 72 +++++++++++++++++++++ 2 files changed, 75 insertions(+), 2 deletions(-) create mode 100644 tests/Unit/V2/EntityReferenceTest.php diff --git a/src/V2/EntityReferences/EntityReference.php b/src/V2/EntityReferences/EntityReference.php index fd75fc5c..4f7238e3 100644 --- a/src/V2/EntityReferences/EntityReference.php +++ b/src/V2/EntityReferences/EntityReference.php @@ -50,8 +50,9 @@ public function toArray(): array */ public static function fromArray(array $value): self { - if (array_keys($value) !== ['type', 'id', 'version'] - || ! is_string($value['type']) || ! is_string($value['id']) || ! is_int($value['version'])) { + if (count($value) !== 3 + || ! is_string($value['type'] ?? null) || ! is_string($value['id'] ?? null) + || ! is_int($value['version'] ?? null)) { throw new InvalidArgumentException('A recorded entity reference must be exactly {type, id, version}.'); } diff --git a/tests/Unit/V2/EntityReferenceTest.php b/tests/Unit/V2/EntityReferenceTest.php new file mode 100644 index 00000000..ad05c5c6 --- /dev/null +++ b/tests/Unit/V2/EntityReferenceTest.php @@ -0,0 +1,72 @@ + 1, + 'id' => 'order-1', + 'type' => 'orders.order', + ]); + + self::assertSame('orders.order', $reference->type); + self::assertSame('order-1', $reference->id); + self::assertSame(1, $reference->version); + self::assertSame([ + 'type' => 'orders.order', + 'id' => 'order-1', + 'version' => 1, + ], $reference->toArray()); + } + + public static function invalidReferences(): iterable + { + yield 'missing type' => [[ + 'id' => 'order-1', + 'version' => 1, + ]]; + yield 'extra field' => [[ + 'type' => 'orders.order', + 'id' => 'order-1', + 'version' => 1, + 'state' => 'paid', + ]]; + yield 'unknown field replaces type' => [[ + 'name' => 'orders.order', + 'id' => 'order-1', + 'version' => 1, + ]]; + yield 'string version' => [[ + 'type' => 'orders.order', + 'id' => 'order-1', + 'version' => '1', + ]]; + yield 'empty id' => [[ + 'type' => 'orders.order', + 'id' => '', + 'version' => 1, + ]]; + yield 'invalid version' => [[ + 'type' => 'orders.order', + 'id' => 'order-1', + 'version' => 0, + ]]; + } + + #[DataProvider('invalidReferences')] + public function testMalformedReferencesRemainRefused(array $value): void + { + $this->expectException(InvalidArgumentException::class); + EntityReference::fromArray($value); + } +} From 9deb878775cd4c2c6d95eac13c204314f9041ecc Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:13:34 +0000 Subject: [PATCH 3/4] Resolve queued activity entity references outside claim transactions --- CHANGELOG.md | 8 + docs/api-stability.md | 7 + docs/entity-references.md | 109 +++++++ scripts/ci/validate-regression-corpus.py | 6 + .../EntityReferenceAdapter.php | 2 + .../EntityReferenceContext.php | 3 +- .../EntityReferenceRefused.php | 11 +- src/V2/Jobs/RunActivityTask.php | 2 + src/V2/Support/ActivityRecovery.php | 4 + src/V2/Support/ActivitySnapshot.php | 6 + src/V2/Support/EmbeddedV2HistoryImport.php | 4 + src/V2/Support/EntityReferenceArguments.php | 272 ++++++++++++++++++ src/V2/Support/HistoryExport.php | 6 + src/V2/Support/LocalActivityExecutor.php | 18 +- src/V2/Support/RunActivityView.php | 3 + src/V2/Support/WorkflowExecutor.php | 14 +- src/config/workflows.php | 1 + .../V2EmbeddedReplayRegressionCorpusTest.php | 44 +++ tests/Feature/V2/V2EntityReferenceTest.php | 145 +++++++++- .../OrderInjectedActivity.php | 17 ++ .../V2/EntityReferences/OrderMapActivity.php | 15 + .../OrderMismatchActivity.php | 15 + .../OrderReferenceResolver.php | 6 + .../V2/EntityReferences/OrderRepository.php | 9 + .../EntityReferences/OrderRetryActivity.php | 25 ++ .../EntityReferences/OrderStatusWorkflow.php | 23 +- ...ntity-reference-attempt-current-state.json | 47 +++ .../Unit/V2/EntityReferenceArgumentsTest.php | 159 ++++++++++ 28 files changed, 964 insertions(+), 17 deletions(-) create mode 100644 docs/entity-references.md create mode 100644 src/V2/Support/EntityReferenceArguments.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderInjectedActivity.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderMapActivity.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderMismatchActivity.php create mode 100644 tests/Fixtures/V2/EntityReferences/OrderRetryActivity.php create mode 100644 tests/Fixtures/V2/ReplayRegression/entity-reference-attempt-current-state.json create mode 100644 tests/Unit/V2/EntityReferenceArgumentsTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 70f78f89..02cebe96 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +- Experimental domain entity references for top-level arguments of embedded + queued activities (#996). Register an entity class, deterministic + adapter and authorized resolver. History records a portable reference and + argument positions. Each real attempt resolves current state outside the + engine transaction; replay never calls the resolver. Ordinary maps and + unregistered objects retain their existing serialization behavior. See + [the registration and retry contract](docs/entity-references.md). + - A run no longer leaves work running after its own code closes it (#1117). Before, an activity or timer the workflow never awaited, such as a `select()` loser, stayed open when the run completed, failed or continued as diff --git a/docs/api-stability.md b/docs/api-stability.md index d879db44..b47333ed 100644 --- a/docs/api-stability.md +++ b/docs/api-stability.md @@ -3,6 +3,13 @@ This document describes the v2 public API stability contract of the `durable-workflow/workflow` package. +The [domain entity reference API](entity-references.md) is experimental. Its +adapter/resolver registrations apply only to top-level arguments of in-process +queued PHP activities. The optional `entity_reference_positions` in +an activity snapshot identifies arguments adapted to portable `{type,id,version}` +maps; absence preserves ordinary argument behavior. History recovery and export +retain this discriminator. It does not advertise a service-mode capability. + ## Relationship to the platform compatibility authority This document is the per-package stability contract for the PHP workflow diff --git a/docs/entity-references.md b/docs/entity-references.md new file mode 100644 index 00000000..b5fb8873 --- /dev/null +++ b/docs/entity-references.md @@ -0,0 +1,109 @@ +# Domain entity references in activities + +This experimental embedded API lets queued PHP activities receive a +domain entity loaded by the application. Workflow history records its logical +type, identifier and reference version. The entity class implements no Workflow +or Laravel interface. + +Choose the argument that expresses the activity's contract: + +- Pass an identifier when the activity should own the lookup explicitly. +- Pass an Avro-compatible value when the activity needs the exact recorded state. +- Pass a registered entity when each attempt should load its current state. + +Only top-level activity arguments are adapted. Nested objects, collections, +activity results and workflow arguments keep their existing serialization rules. +This API does not add object serialization or a Server/SDK capability. + +## Register the application boundary + +Configure an entity class, adapter and resolver under a stable logical type: + +```php +// config/workflows.php +'v2' => [ + 'entity_references' => [ + 'orders.order' => [ + 'entity_class' => App\Domain\Order::class, + 'adapter' => App\Infrastructure\OrderReferenceAdapter::class, + 'resolver' => App\Infrastructure\OrderReferenceResolver::class, + ], + ], +], +``` + +The class binding lets the worker check the declared activity parameter before +repository access and preserve Laravel method injection. Class names stay in +application configuration. The recorded reference contains only +`{type: "orders.order", id: "o-123", version: 1}` and explicit argument positions; +an ordinary map with the same keys remains an ordinary value. + +The adapter implements `Workflow\V2\EntityReferences\EntityReferenceAdapter`: + +```php +final class OrderReferenceAdapter implements EntityReferenceAdapter +{ + public function supports(object $entity): bool + { + return $entity instanceof Order; + } + + public function identify(object $entity): string + { + return $entity->id; + } + + public function version(): int + { + return 1; + } +} +``` + +Adapter construction and methods must be deterministic and perform no I/O. +Adaptation runs while the workflow turn is committed and may run again on a +transaction retry. Workflow code must follow its usual replay rules; registering +an adapter does not permit database reads inside workflow code. + +The resolver implements `EntityReferenceResolver`. It receives the portable +reference and an engine-supplied `EntityReferenceContext` containing namespace, +workflow instance/run, activity execution, attempt number, argument position and +the named object parameter type when one is declared: + +```php +public function resolve(EntityReference $reference, EntityReferenceContext $context): object +{ + $order = $this->orders->find($reference->id) + ?? throw EntityNotFound::for($reference); + if (! $this->authorization->canProcess($context->namespace, $order)) { + throw EntityAccessDenied::for($reference); + } + return $order; +} +``` + +The resolver owns application authorization. A reference identifier is not proof +of ownership or tenancy. Resolution happens outside the engine's claim/commit +transaction before the actual activity method runs. Each retry resolves again; +replay consumes the recorded activity outcome without calling the resolver. +Other Laravel dependencies are still injected normally. + +```php +public function handle(LoggerInterface $logger, Order $order): void +{ + // $order is the authorized current entity for this attempt. +} +``` + +Unknown logical types, unsupported reference versions, malformed metadata and +parameter mismatches fail before the method runs with non-retryable +`EntityReferenceRefused`. Missing entities, denied access and resolver exceptions +follow the activity's ordinary retry policy. Deploy matching registrations to +the in-process workers that can receive the activity, and keep them available +for retained history and retries. + +Local entity arguments are refused before scheduling or lookup in this increment. +Local callbacks still execute inside the hosting workflow transaction. Supporting +a resolver there requires splitting preparation, callback and outcome recording +so application I/O runs outside that transaction. That accepted follow-up remains +open in [issue 996](https://github.com/durable-workflow/workflow/issues/996). diff --git a/scripts/ci/validate-regression-corpus.py b/scripts/ci/validate-regression-corpus.py index b1a5ba97..b9d9453c 100644 --- a/scripts/ci/validate-regression-corpus.py +++ b/scripts/ci/validate-regression-corpus.py @@ -158,6 +158,12 @@ ) ]: ( "tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php", + "tests/Fixtures/V2/EntityReferences/Order.php", + "tests/Fixtures/V2/EntityReferences/OrderReferenceAdapter.php", + "tests/Fixtures/V2/EntityReferences/OrderReferenceResolver.php", + "tests/Fixtures/V2/EntityReferences/OrderRepository.php", + "tests/Fixtures/V2/EntityReferences/OrderStatusActivity.php", + "tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php", "tests/Fixtures/V2/TestConstructorInjectionActivity.php", "tests/Fixtures/V2/TestConstructorInjectionWorkflow.php", "tests/Fixtures/V2/TestTracingReplayActivity.php", diff --git a/src/V2/EntityReferences/EntityReferenceAdapter.php b/src/V2/EntityReferences/EntityReferenceAdapter.php index 938bf095..4d4c2e57 100644 --- a/src/V2/EntityReferences/EntityReferenceAdapter.php +++ b/src/V2/EntityReferences/EntityReferenceAdapter.php @@ -11,6 +11,8 @@ * Registered by logical type in `workflows.v2.entity_references`. The domain * class itself implements nothing from Durable Workflow or Laravel; the * adapter belongs to the application's infrastructure layer. + * Its constructor and methods must be deterministic and perform no I/O: + * adaptation happens while the workflow turn is committed, including retries. * * @experimental Draft API for workflow issue 996; not covered by the stability * guarantees until it ships. diff --git a/src/V2/EntityReferences/EntityReferenceContext.php b/src/V2/EntityReferences/EntityReferenceContext.php index 47a2ef34..c531e402 100644 --- a/src/V2/EntityReferences/EntityReferenceContext.php +++ b/src/V2/EntityReferences/EntityReferenceContext.php @@ -14,7 +14,8 @@ final class EntityReferenceContext { /** * @param class-string|null $parameterType The declared parameter type the - * resolved object must satisfy, or null for an untyped parameter. + * resolved object must satisfy, or null for an untyped, built-in or + * compound type. The worker also checks compound types locally. */ public function __construct( public readonly ?string $namespace, diff --git a/src/V2/EntityReferences/EntityReferenceRefused.php b/src/V2/EntityReferences/EntityReferenceRefused.php index 93899b33..1c71b0be 100644 --- a/src/V2/EntityReferences/EntityReferenceRefused.php +++ b/src/V2/EntityReferences/EntityReferenceRefused.php @@ -5,6 +5,7 @@ namespace Workflow\V2\EntityReferences; use RuntimeException; +use Workflow\Exceptions\NonRetryableExceptionContract; /** * The worker refused a recorded reference before the activity ran (#996): @@ -13,8 +14,16 @@ * * @experimental Draft API for workflow issue 996. */ -final class EntityReferenceRefused extends RuntimeException +final class EntityReferenceRefused extends RuntimeException implements NonRetryableExceptionContract { + public const INVALID_REFERENCE = 'invalid_entity_reference'; + + public const INVALID_REGISTRATION = 'invalid_entity_reference_registration'; + + public const AMBIGUOUS_TYPE = 'ambiguous_entity_reference_type'; + + public const UNSUPPORTED_EXECUTION_MODE = 'entity_reference_execution_mode_not_supported'; + public const UNKNOWN_TYPE = 'unknown_entity_reference_type'; public const UNSUPPORTED_VERSION = 'unsupported_entity_reference_version'; diff --git a/src/V2/Jobs/RunActivityTask.php b/src/V2/Jobs/RunActivityTask.php index 5c93311f..ed0356a5 100644 --- a/src/V2/Jobs/RunActivityTask.php +++ b/src/V2/Jobs/RunActivityTask.php @@ -18,6 +18,7 @@ use Workflow\V2\Support\ActivityTaskClaim; use Workflow\V2\Support\ActivityTaskClaimer; use Workflow\V2\Support\EmbeddedActivityLease; +use Workflow\V2\Support\EntityReferenceArguments; use Workflow\V2\Support\EntryMethod; use Workflow\V2\Support\ExecutionTracing; use Workflow\V2\Support\NamespaceScope; @@ -135,6 +136,7 @@ private function executeMaintainedClaim(ActivityTaskClaim $claim, ?EmbeddedActiv $result = null; $throwable = null; try { + $activityArguments = EntityReferenceArguments::resolve($activity, $activityArguments, $entryMethod); $arguments = ActivityEntryArguments::resolve($activity, $activityArguments, $entryMethod); } catch (Throwable $error) { $arguments = []; diff --git a/src/V2/Support/ActivityRecovery.php b/src/V2/Support/ActivityRecovery.php index 208fb2ef..45ef8b18 100644 --- a/src/V2/Support/ActivityRecovery.php +++ b/src/V2/Support/ActivityRecovery.php @@ -98,6 +98,10 @@ public static function restore(WorkflowRun $run, string $activityId): ?ActivityE 'queue' => self::stringValue($state['queue'] ?? null) ?? $run->queue, 'attempt_count' => max(0, self::intValue($state['attempt_count'] ?? null) ?? 0), 'retry_policy' => is_array($state['retry_policy'] ?? null) ? $state['retry_policy'] : null, + 'activity_options' => isset($state[EntityReferenceArguments::POSITIONS]) + ? [ + EntityReferenceArguments::POSITIONS => $state[EntityReferenceArguments::POSITIONS], + ] : null, 'parallel_group_path' => is_array($state['parallel_group_path'] ?? null) ? $state['parallel_group_path'] : null, diff --git a/src/V2/Support/ActivitySnapshot.php b/src/V2/Support/ActivitySnapshot.php index ef5d04c6..e0fb2e9f 100644 --- a/src/V2/Support/ActivitySnapshot.php +++ b/src/V2/Support/ActivitySnapshot.php @@ -36,6 +36,9 @@ public static function fromExecution(ActivityExecution $execution): array 'attempt_count' => self::executionAttemptCount($execution), 'retry_policy' => self::arrayValue($execution->retry_policy), 'worker_session' => self::arrayValue($execution->activity_options['worker_session'] ?? null), + EntityReferenceArguments::POSITIONS => self::arrayValue( + $execution->activity_options[EntityReferenceArguments::POSITIONS] ?? null + ), 'cancellation_policy' => self::stringValue($execution->activity_options['cancellation_policy'] ?? null), 'cancellation_scope_id' => self::stringValue($execution->activity_options['cancellation_scope_id'] ?? null), 'schedule_to_close_deadline_at' => ($execution->activity_options['cancellation_policy'] ?? null) @@ -189,6 +192,9 @@ private static function sanitizeSnapshot(array $snapshot): array 'attempt_count' => self::intValue($snapshot['attempt_count'] ?? null), 'retry_policy' => self::arrayValue($snapshot['retry_policy'] ?? null), 'worker_session' => self::arrayValue($snapshot['worker_session'] ?? null), + EntityReferenceArguments::POSITIONS => self::arrayValue( + $snapshot[EntityReferenceArguments::POSITIONS] ?? null + ), 'cancellation_policy' => self::stringValue($snapshot['cancellation_policy'] ?? null), 'cancellation_scope_id' => self::stringValue($snapshot['cancellation_scope_id'] ?? null), 'schedule_to_close_deadline_at' => self::stringValue($snapshot['schedule_to_close_deadline_at'] ?? null), diff --git a/src/V2/Support/EmbeddedV2HistoryImport.php b/src/V2/Support/EmbeddedV2HistoryImport.php index 3b5590eb..859a298c 100644 --- a/src/V2/Support/EmbeddedV2HistoryImport.php +++ b/src/V2/Support/EmbeddedV2HistoryImport.php @@ -741,6 +741,10 @@ private static function writeActivities(array $bundle, string $runId, string $ru 'attempt_count' => self::intValue($activity['attempt_count'] ?? null) ?? 0, 'current_attempt_id' => self::stringValue($activity['current_attempt_id'] ?? null), 'retry_policy' => self::arrayValue($activity['retry_policy'] ?? null) ?? [], + 'activity_options' => isset($activity[EntityReferenceArguments::POSITIONS]) + ? [ + EntityReferenceArguments::POSITIONS => $activity[EntityReferenceArguments::POSITIONS], + ] : null, 'parallel_group_path' => self::arrayValue($activity['parallel_group_path'] ?? null) ?? [], 'started_at' => self::timestamp($activity['started_at'] ?? null), 'closed_at' => self::timestamp($activity['closed_at'] ?? null), diff --git a/src/V2/Support/EntityReferenceArguments.php b/src/V2/Support/EntityReferenceArguments.php new file mode 100644 index 00000000..3de1afa5 --- /dev/null +++ b/src/V2/Support/EntityReferenceArguments.php @@ -0,0 +1,272 @@ +} + */ + public static function encode(array $arguments): array + { + $positions = []; + foreach ($arguments as $position => $value) { + if (! is_object($value)) { + continue; + } + $reference = null; + foreach (self::registrations() as $type => $definition) { + $definition = self::definition($type, $definition); + $adapter = app() + ->make($definition['adapter']); + if (! $value instanceof $definition['entity_class'] || ! $adapter->supports($value)) { + continue; + } + if ($reference !== null) { + throw new EntityReferenceRefused( + EntityReferenceRefused::AMBIGUOUS_TYPE, + 'More than one entity type adapts this activity argument.' + ); + } + $reference = new EntityReference($type, $adapter->identify($value), $adapter->version()); + } + if ($reference !== null) { + if (! is_int($position) || ! array_is_list($arguments)) { + throw new EntityReferenceRefused( + EntityReferenceRefused::INVALID_REFERENCE, + 'Entity references require positional activity arguments.' + ); + } + $arguments[$position] = $reference->toArray(); + $positions[] = $position; + } + } + return [$arguments, $positions]; + } + + public static function resolve(Activity $activity, array $arguments, ReflectionMethod $entry): array + { + $options = $activity->execution->activity_options ?? []; + if (! is_array($options) || ! array_key_exists(self::POSITIONS, $options)) { + return $arguments; + } + $positions = $options[self::POSITIONS]; + if ($positions === []) { + return $arguments; + } + if (! is_array($positions) || ! array_is_list($positions) || ! array_is_list($arguments)) { + throw self::invalid(); + } + $references = []; + $definitions = []; + foreach ($positions as $position) { + if (! is_int($position) || $position < 0 || isset($references[$position]) + || ! isset($arguments[$position]) || ! is_array($arguments[$position])) { + throw self::invalid(); + } + try { + $reference = EntityReference::fromArray($arguments[$position]); + } catch (InvalidArgumentException) { + throw self::invalid(); + } + $definition = self::definition($reference->type, self::registrations()[$reference->type] ?? null); + $adapter = app() + ->make($definition['adapter']); + if ($reference->version !== $adapter->version()) { + throw new EntityReferenceRefused( + EntityReferenceRefused::UNSUPPORTED_VERSION, + 'The recorded entity reference version is not supported.' + ); + } + $references[$position] = $reference; + $definitions[$position] = $definition; + } + $parameters = self::parameters($arguments, $definitions, $entry); + foreach ($references as $position => $reference) { + $definition = $definitions[$position]; + $parameter = $parameters[$position]; + $parameterType = $parameter->getType(); + $declaredType = $parameterType instanceof ReflectionNamedType && ! $parameterType->isBuiltin() + ? $parameterType->getName() : null; + $context = new EntityReferenceContext( + namespace: $activity->run->namespace, + workflowInstanceId: $activity->run->workflow_instance_id, + workflowRunId: $activity->run->id, + activityExecutionId: $activity->execution->id, + attemptNumber: (int) $activity->execution->attempt_count, + argumentPosition: $position, + parameterType: $declaredType, + ); + $value = app() + ->make($definition['resolver'])->resolve($reference, $context); + if (! $value instanceof $definition['entity_class'] || ! self::accepts( + $parameterType, + $value::class, + $parameter + )) { + throw self::mismatch(); + } + $arguments[$position] = $value; + } + return $arguments; + } + + /** + * @return array + */ + private static function parameters(array $arguments, array $definitions, ReflectionMethod $entry): array + { + $result = []; + $position = 0; + $classes = []; + foreach ($arguments as $index => $value) { + $classes[] = $definitions[$index]['entity_class'] ?? (is_object($value) ? $value::class : null); + } + foreach ($entry->getParameters() as $parameter) { + if (! array_key_exists($position, $arguments)) { + break; + } + $type = $parameter->getType(); + // Match the existing method-injection rule: an unmatched class + // parameter is injected without consuming recorded input. The + // method resolver checks all arguments, including later ones. + $injected = $parameter->getAttributes( + 'Illuminate\Contracts\Container\ContextualAttribute', + \ReflectionAttribute::IS_INSTANCEOF + ) !== []; + if ($type instanceof ReflectionNamedType && ! $type->isBuiltin()) { + $provided = false; + foreach ($classes as $class) { + $provided = $provided || ($class !== null && self::accepts($type, $class, $parameter)); + } + $injected = $injected || ! $provided; + } + if ($injected) { + continue; + } + do { + if (isset($definitions[$position])) { + if (! self::accepts($type, $definitions[$position]['entity_class'], $parameter)) { + throw self::mismatch(); + } + $result[$position] = $parameter; + } + ++$position; + } while ($parameter->isVariadic() && array_key_exists($position, $arguments)); + } + if (count($result) !== count($definitions)) { + throw self::mismatch(); + } + return $result; + } + + private static function accepts(?ReflectionType $type, string $class, ReflectionParameter $parameter): bool + { + if ($type === null) { + return true; + } + if ($type instanceof ReflectionUnionType) { + foreach ($type->getTypes() as $part) { + if (self::accepts($part, $class, $parameter)) { + return true; + } + } + return false; + } + if ($type instanceof ReflectionIntersectionType) { + foreach ($type->getTypes() as $part) { + if (! self::accepts($part, $class, $parameter)) { + return false; + } + } + return true; + } + $name = $type instanceof ReflectionNamedType ? $type->getName() : ''; + $declaring = $parameter->getDeclaringClass(); + if ($name === 'self') { + $name = $declaring?->getName() ?? $name; + } elseif ($name === 'parent') { + $parent = $declaring?->getParentClass(); + $name = $parent ? $parent->getName() : $name; + } + return $type instanceof ReflectionNamedType && ($type->isBuiltin() + ? in_array($type->getName(), ['mixed', 'object'], true) + : is_a($class, $name, true)); + } + + private static function registrations(): array + { + $definitions = config('workflows.v2.entity_references', []); + if (! is_array($definitions)) { + throw new EntityReferenceRefused( + EntityReferenceRefused::INVALID_REGISTRATION, + 'Entity reference registrations must be a type map.' + ); + } + return $definitions; + } + + private static function definition(mixed $type, mixed $definition): array + { + if ($definition === null) { + throw new EntityReferenceRefused( + EntityReferenceRefused::UNKNOWN_TYPE, + 'The recorded entity type is not registered.' + ); + } + if (! is_string($type) || $type === '' || ! is_array($definition) + || ! is_string($definition['entity_class'] ?? null) + || (! class_exists($definition['entity_class']) && ! interface_exists($definition['entity_class'])) + || ! is_string($definition['adapter'] ?? null) || ! is_a( + $definition['adapter'], + EntityReferenceAdapter::class, + true + ) + || ! is_string($definition['resolver'] ?? null) || ! is_a( + $definition['resolver'], + EntityReferenceResolver::class, + true + )) { + throw new EntityReferenceRefused( + EntityReferenceRefused::INVALID_REGISTRATION, + 'An entity type needs its entity_class, adapter and resolver registrations.' + ); + } + return $definition; + } + + private static function invalid(): EntityReferenceRefused + { + return new EntityReferenceRefused( + EntityReferenceRefused::INVALID_REFERENCE, + 'Recorded entity reference metadata is invalid.' + ); + } + + private static function mismatch(): EntityReferenceRefused + { + return new EntityReferenceRefused( + EntityReferenceRefused::PARAMETER_MISMATCH, + 'The registered entity type does not match its activity parameter.' + ); + } +} diff --git a/src/V2/Support/HistoryExport.php b/src/V2/Support/HistoryExport.php index 608f39e8..d55a02cd 100644 --- a/src/V2/Support/HistoryExport.php +++ b/src/V2/Support/HistoryExport.php @@ -1510,6 +1510,9 @@ private static function activityState(array $activity, array $state = []): array 'connection' => $activity['connection'] ?? ($state['connection'] ?? null), 'queue' => $activity['queue'] ?? ($state['queue'] ?? null), 'retry_policy' => $state['retry_policy'] ?? ($activity['retry_policy'] ?? null), + ...(isset($state[EntityReferenceArguments::POSITIONS]) ? [ + EntityReferenceArguments::POSITIONS => $state[EntityReferenceArguments::POSITIONS], + ] : []), 'attempt_count' => self::intValue($activity['attempt_count'] ?? null) ?? self::intValue($state['attempt_count'] ?? null) ?? 0, @@ -1639,6 +1642,9 @@ private static function activity(array $activity): array 'connection' => $activity['connection'] ?? null, 'queue' => $activity['queue'] ?? null, 'retry_policy' => $activity['retry_policy'] ?? null, + ...(isset($activity[EntityReferenceArguments::POSITIONS]) ? [ + EntityReferenceArguments::POSITIONS => $activity[EntityReferenceArguments::POSITIONS], + ] : []), 'attempt_count' => $activity['attempt_count'] ?? 0, 'current_attempt_id' => $activity['current_attempt_id'] ?? null, 'current_worker_attempt_id' => $activity['current_worker_attempt_id'] ?? null, diff --git a/src/V2/Support/LocalActivityExecutor.php b/src/V2/Support/LocalActivityExecutor.php index 0940f542..1f13793d 100644 --- a/src/V2/Support/LocalActivityExecutor.php +++ b/src/V2/Support/LocalActivityExecutor.php @@ -12,6 +12,7 @@ use Throwable; use Workflow\Serializers\CodecRegistry; use Workflow\Serializers\Serializer; +use Workflow\V2\EntityReferences\EntityReferenceRefused; use Workflow\V2\Enums\ActivityAttemptStatus; use Workflow\V2\Enums\ActivityStatus; use Workflow\V2\Enums\FailureCategory; @@ -626,6 +627,12 @@ private function invokeCallback(WorkflowRun $run, WorkflowTask $task, ActivityEx $result = null; $throwable = null; try { + if (array_key_exists(EntityReferenceArguments::POSITIONS, $execution->activity_options ?? [])) { + throw new EntityReferenceRefused( + EntityReferenceRefused::UNSUPPORTED_EXECUTION_MODE, + 'Local entity references require resolution outside the hosting workflow transaction.', + ); + } $arguments = ActivityEntryArguments::resolve($activity, $activityArguments, $entryMethod); } catch (Throwable $error) { $arguments = []; @@ -938,8 +945,15 @@ private function createExecution( ? now() ->addSeconds($options->scheduleToCloseTimeout) : null; - $argumentsCodec = Serializer::chooseCodecForData($run->payload_codec, $activityCall->arguments); - $serializedArguments = Serializer::serializeWithCodec($argumentsCodec, $activityCall->arguments); + [$arguments, $referencePositions] = EntityReferenceArguments::encode($activityCall->arguments); + if ($referencePositions !== []) { + throw new EntityReferenceRefused( + EntityReferenceRefused::UNSUPPORTED_EXECUTION_MODE, + 'Local entity references require resolution outside the hosting workflow transaction.', + ); + } + $argumentsCodec = Serializer::chooseCodecForData($run->payload_codec, $arguments); + $serializedArguments = Serializer::serializeWithCodec($argumentsCodec, $arguments); $storedArguments = ExternalPayloads::externalizeForNamespace( $serializedArguments, $argumentsCodec, diff --git a/src/V2/Support/RunActivityView.php b/src/V2/Support/RunActivityView.php index 79dd936d..f32b18e5 100644 --- a/src/V2/Support/RunActivityView.php +++ b/src/V2/Support/RunActivityView.php @@ -294,6 +294,9 @@ private static function presentActivity( 'status' => $status, 'attempt_count' => $attemptCount, 'retry_policy' => $state['retry_policy'] ?? ($execution?->retry_policy ?? null), + ...(isset($state[EntityReferenceArguments::POSITIONS]) ? [ + EntityReferenceArguments::POSITIONS => $state[EntityReferenceArguments::POSITIONS], + ] : []), 'connection' => $state['connection'] ?? null, 'queue' => $state['queue'] ?? null, 'last_heartbeat_at' => $state['last_heartbeat_at'] ?? ($latestAttempt['last_heartbeat_at'] ?? null), diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index 4b028224..2aa2c814 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -2530,10 +2530,9 @@ private function scheduleActivity( ->addSeconds($options->scheduleToCloseTimeout) : null; - // Persist the sole v2 codec beside the activity row. Callers must adapt - // PHP-only objects to values representable by the fixed Avro schema. - $argumentsCodec = Serializer::chooseCodecForData($run->payload_codec, $activityCall->arguments); - $serializedArguments = Serializer::serializeWithCodec($argumentsCodec, $activityCall->arguments); + [$arguments, $referencePositions] = EntityReferenceArguments::encode($activityCall->arguments); + $argumentsCodec = Serializer::chooseCodecForData($run->payload_codec, $arguments); + $serializedArguments = Serializer::serializeWithCodec($argumentsCodec, $arguments); $this->logApproachingLimit( StructuralLimits::warnApproachingPayloadSize($serializedArguments), $run, @@ -2562,7 +2561,12 @@ private function scheduleActivity( 'connection' => RoutingResolver::activityConnection($activityCall->activity, $run, $options), 'queue' => RoutingResolver::activityQueue($activityCall->activity, $run, $options), 'parallel_group_path' => self::parallelGroupPath($parallelMetadata), - 'activity_options' => $options?->toSnapshot(), + 'activity_options' => $referencePositions === [] ? $options?->toSnapshot() : array_merge( + $options?->toSnapshot() ?? [], + [ + EntityReferenceArguments::POSITIONS => $referencePositions, + ], + ), 'schedule_deadline_at' => $scheduleDeadlineAt, 'schedule_to_close_deadline_at' => $scheduleToCloseDeadlineAt, ]); diff --git a/src/config/workflows.php b/src/config/workflows.php index 159d13d2..ad566813 100644 --- a/src/config/workflows.php +++ b/src/config/workflows.php @@ -125,6 +125,7 @@ // attempt. Top-level activity arguments of in-process workers only. 'entity_references' => [ // 'orders.order' => [ + // 'entity_class' => App\Domain\Order::class, // 'adapter' => App\Infrastructure\OrderReferenceAdapter::class, // 'resolver' => App\Infrastructure\OrderReferenceResolver::class, // ], diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index 9becd71e..d32b27ca 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -121,6 +121,50 @@ final class V2EmbeddedReplayRegressionCorpusTest extends TestCase private int $workflowNumber = 0; + public function testEntityReferenceAttemptCorpusLoadsCurrentStateAndNeverLoadsOnReplay(): void + { + $path = self::FIXTURE_DIR . '/entity-reference-attempt-current-state.json'; + if (! is_file($path)) { + $this->markTestSkipped('The entity reference attempt evidence is absent.'); + } + $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR); + $scenario = $fixture['entity_reference_attempt']; + self::stopWorkers(); + Queue::fake(); + config([ + 'queue.default' => 'database', + 'workflows.v2.entity_references' => [ + $scenario['logical_type'] => [ + 'entity_class' => \Tests\Fixtures\V2\EntityReferences\Order::class, + 'adapter' => \Tests\Fixtures\V2\EntityReferences\OrderReferenceAdapter::class, + 'resolver' => \Tests\Fixtures\V2\EntityReferences\OrderReferenceResolver::class, + ], + ], + ]); + \Tests\Fixtures\V2\EntityReferences\OrderRepository::reset(); + $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-entity-reference'); + $stub->start(...$fixture['workflow']['arguments']); + $runId = $stub->runId(); + $this->runReadyTaskOfType($runId, TaskType::Workflow); + $execution = ActivityExecution::query()->where('workflow_run_id', $runId)->sole(); + $this->assertSame($scenario['recorded_positions'], $execution->activity_options['entity_reference_positions']); + $this->assertSame([ + 'type' => $scenario['logical_type'], + 'id' => $scenario['entity_id'], + 'version' => 1, + ], $execution->activityArguments()[0]); + \Tests\Fixtures\V2\EntityReferences\OrderRepository::save(new \Tests\Fixtures\V2\EntityReferences\Order( + $scenario['entity_id'], + $scenario['state_at_attempt'], + )); + $this->runReadyTaskOfType($runId, TaskType::Activity); + $this->runReadyTaskOfType($runId, TaskType::Workflow); + $this->assertSame($scenario['expected_result'], $stub->refresh()->output()); + (new WorkflowReplayer())->replay(WorkflowRun::query()->findOrFail($runId)); + $this->assertSame($scenario['loads_after_replay'], \Tests\Fixtures\V2\EntityReferences\OrderRepository::$loads); + $this->assertSame([0], \Tests\Fixtures\V2\EntityReferences\OrderRepository::$transactionLevels); + } + public function testImportedChildLineageCorpusDoesNotAttachAnUnrelatedLaterRun(): void { $path = self::FIXTURE_DIR . '/imported-child-lineage-requires-owned-edge.json'; diff --git a/tests/Feature/V2/V2EntityReferenceTest.php b/tests/Feature/V2/V2EntityReferenceTest.php index 82d4d1d3..2fad648a 100644 --- a/tests/Feature/V2/V2EntityReferenceTest.php +++ b/tests/Feature/V2/V2EntityReferenceTest.php @@ -12,6 +12,7 @@ use Tests\Fixtures\V2\EntityReferences\OrderStatusWorkflow; use Tests\TestCase; use Workflow\Serializers\Serializer; +use Workflow\V2\EntityReferences\EntityAccessDenied; use Workflow\V2\EntityReferences\EntityNotFound; use Workflow\V2\EntityReferences\EntityReferenceRefused; use Workflow\V2\Enums\HistoryEventType; @@ -23,13 +24,11 @@ use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Support\ActivityRecovery; +use Workflow\V2\Support\HistoryExport; use Workflow\V2\Support\WorkflowReplayer; use Workflow\V2\WorkflowStub; -/** - * Domain entity references in activity arguments (#996). Draft: these pin the - * accepted behavior and fail until the adapter/resolver implementation lands. - */ final class V2EntityReferenceTest extends TestCase { protected function setUp(): void @@ -41,6 +40,7 @@ protected function setUp(): void config() ->set('workflows.v2.entity_references', [ 'orders.order' => [ + 'entity_class' => Order::class, 'adapter' => OrderReferenceAdapter::class, 'resolver' => OrderReferenceResolver::class, ], @@ -71,12 +71,23 @@ public function testAnActivityReceivesTheCurrentEntityResolvedFromAPortableRefer $this->assertSame([0], $execution->activity_options['entity_reference_positions'] ?? null); $this->assertSame([0], $scheduled->payload['activity']['entity_reference_positions'] ?? null); + OrderRepository::save(new Order('o-1', 'shipped')); $this->runReadyTask($runId, TaskType::Activity); $this->runReadyTask($runId, TaskType::Workflow); // The activity saw the persisted state, not the workflow's copy. - $this->assertSame('o-1:paid:checked', $workflow->refresh()->output()); + $this->assertSame('o-1:shipped:checked', $workflow->refresh()->output()); $this->assertSame(1, OrderRepository::$loads); + $context = OrderRepository::$contexts[0]; + $run = WorkflowRun::query()->findOrFail($runId); + $this->assertSame($run->namespace, $context->namespace); + $this->assertSame($run->workflow_instance_id, $context->workflowInstanceId); + $this->assertSame($runId, $context->workflowRunId); + $this->assertSame($execution->id, $context->activityExecutionId); + $this->assertSame(1, $context->attemptNumber); + $this->assertSame(0, $context->argumentPosition); + $this->assertSame(Order::class, $context->parameterType); + $this->assertSame([0], OrderRepository::$transactionLevels); } public function testReplayNeverResolvesTheReference(): void @@ -148,6 +159,130 @@ public function testAnUnregisteredObjectStillFailsWithoutObjectSerialization(): ->exists()); } + public function testAnOrdinaryReferenceShapedMapIsNeverResolved(): void + { + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-map'); + $workflow->start('o-map', 'map'); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $execution = ActivityExecution::query()->where('workflow_run_id', $workflow->runId())->sole(); + $this->assertArrayNotHasKey('entity_reference_positions', $execution->activity_options ?? []); + $this->runReadyTask($workflow->runId(), TaskType::Activity); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->assertSame('o-map:checked', $workflow->refresh()->output()); + $this->assertSame(0, OrderRepository::$loads); + } + + public function testEntityResolutionPreservesMethodDependencyInjection(): void + { + OrderRepository::save(new Order('o-di', 'paid')); + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-di'); + $workflow->start('o-di', 'injected'); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->runReadyTask($workflow->runId(), TaskType::Activity); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->assertSame('o-di:paid:checked', $workflow->refresh()->output()); + $this->assertSame(Order::class, OrderRepository::$contexts[0]->parameterType); + } + + public function testAParameterMismatchIsRefusedBeforeRepositoryAccess(): void + { + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-mismatch'); + $workflow->start('o-mismatch', 'mismatch'); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->runReadyTask($workflow->runId(), TaskType::Activity); + $failed = WorkflowHistoryEvent::query()->where('workflow_run_id', $workflow->runId()) + ->where('event_type', HistoryEventType::ActivityFailed->value)->sole(); + $this->assertSame(EntityReferenceRefused::class, $failed->payload['exception_class']); + $this->assertTrue($failed->payload['non_retryable']); + $this->assertSame([], OrderRepository::$contexts); + } + + public function testAnUnsupportedVersionIsRefusedBeforeRepositoryAccess(): void + { + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-version'); + $workflow->start('o-version'); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $execution = ActivityExecution::query()->where('workflow_run_id', $workflow->runId())->sole(); + $arguments = $execution->activityArguments(); + $arguments[0]['version'] = 2; + $execution->forceFill([ + 'arguments' => Serializer::serializeWithCodec($execution->payload_codec, $arguments), + ])->save(); + $this->runReadyTask($workflow->runId(), TaskType::Activity); + $failed = WorkflowHistoryEvent::query()->where('workflow_run_id', $workflow->runId()) + ->where('event_type', HistoryEventType::ActivityFailed->value)->sole(); + $this->assertSame(EntityReferenceRefused::class, $failed->payload['exception_class']); + $this->assertTrue($failed->payload['non_retryable']); + $this->assertSame([], OrderRepository::$contexts); + } + + public function testAuthorizationUsesTheResolverAndNeverEntersTheActivity(): void + { + OrderRepository::$deny = true; + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-denied'); + $workflow->start('o-denied'); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->runReadyTask($workflow->runId(), TaskType::Activity); + $failed = WorkflowHistoryEvent::query()->where('workflow_run_id', $workflow->runId()) + ->where('event_type', HistoryEventType::ActivityFailed->value)->sole(); + $this->assertSame(EntityAccessDenied::class, $failed->payload['exception_class']); + $this->assertSame(0, OrderRepository::$loads); + $this->assertCount(1, OrderRepository::$contexts); + } + + public function testEachRetryLoadsCurrentStateAgain(): void + { + OrderRepository::save(new Order('o-retry', 'paid')); + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-retry'); + $workflow->start('o-retry', 'retry'); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->runReadyTask($workflow->runId(), TaskType::Activity); + OrderRepository::save(new Order('o-retry', 'refunded')); + $this->runReadyTask($workflow->runId(), TaskType::Activity); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->assertSame('o-retry:refunded:checked', $workflow->refresh()->output()); + $this->assertSame( + [1, 2], + array_map(static fn ($context) => $context->attemptNumber, OrderRepository::$contexts) + ); + $this->assertSame([0, 0], OrderRepository::$transactionLevels); + } + + public function testALocalActivityRefusesReferencesBeforeLookupOrScheduling(): void + { + OrderRepository::save(new Order('o-local', 'paid')); + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-local'); + $workflow->start('o-local', 'local'); + try { + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->fail('Local resolution was admitted inside the workflow transaction.'); + } catch (EntityReferenceRefused $error) { + $this->assertSame(EntityReferenceRefused::UNSUPPORTED_EXECUTION_MODE, $error->reason); + } + $this->assertSame([], OrderRepository::$contexts); + $this->assertSame(0, ActivityExecution::query()->where('workflow_run_id', $workflow->runId())->count()); + $this->assertSame(0, WorkflowTask::query()->where('workflow_run_id', $workflow->runId()) + ->where('task_type', TaskType::Activity->value)->count()); + } + + public function testHistoryRecoveryAndExportRetainReferencePositions(): void + { + OrderRepository::save(new Order('o-recovered', 'paid')); + $workflow = WorkflowStub::make(OrderStatusWorkflow::class, 'entity-reference-recovery'); + $workflow->start('o-recovered'); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $execution = ActivityExecution::query()->where('workflow_run_id', $workflow->runId())->sole(); + $id = $execution->id; + $bundle = HistoryExport::forRun(WorkflowRun::query()->findOrFail($workflow->runId())); + $this->assertSame([0], $bundle['activities'][0]['entity_reference_positions'] ?? null); + $execution->delete(); + $restored = ActivityRecovery::restore(WorkflowRun::query()->findOrFail($workflow->runId()), $id); + $this->assertSame([0], $restored?->activity_options['entity_reference_positions'] ?? null); + $this->runReadyTask($workflow->runId(), TaskType::Activity); + $this->runReadyTask($workflow->runId(), TaskType::Workflow); + $this->assertSame('o-recovered:paid:checked', $workflow->refresh()->output()); + } + private function runReadyTask(string $runId, TaskType $taskType): void { $task = WorkflowTask::query() diff --git a/tests/Fixtures/V2/EntityReferences/OrderInjectedActivity.php b/tests/Fixtures/V2/EntityReferences/OrderInjectedActivity.php new file mode 100644 index 00000000..0a1846d8 --- /dev/null +++ b/tests/Fixtures/V2/EntityReferences/OrderInjectedActivity.php @@ -0,0 +1,17 @@ +debug('Order reference activity ran.'); + return $order->id . ':' . $order->status . ':' . $note; + } +} diff --git a/tests/Fixtures/V2/EntityReferences/OrderMapActivity.php b/tests/Fixtures/V2/EntityReferences/OrderMapActivity.php new file mode 100644 index 00000000..49b0b0e0 --- /dev/null +++ b/tests/Fixtures/V2/EntityReferences/OrderMapActivity.php @@ -0,0 +1,15 @@ +id) ?? throw EntityNotFound::for($reference); } } diff --git a/tests/Fixtures/V2/EntityReferences/OrderRepository.php b/tests/Fixtures/V2/EntityReferences/OrderRepository.php index dbb005f8..07d1d9c8 100644 --- a/tests/Fixtures/V2/EntityReferences/OrderRepository.php +++ b/tests/Fixtures/V2/EntityReferences/OrderRepository.php @@ -17,10 +17,19 @@ final class OrderRepository public static int $loads = 0; + public static array $contexts = []; + + public static array $transactionLevels = []; + + public static bool $deny = false; + public static function reset(): void { self::$orders = []; self::$loads = 0; + self::$contexts = []; + self::$transactionLevels = []; + self::$deny = false; } public static function save(Order $order): void diff --git a/tests/Fixtures/V2/EntityReferences/OrderRetryActivity.php b/tests/Fixtures/V2/EntityReferences/OrderRetryActivity.php new file mode 100644 index 00000000..a9dfad1b --- /dev/null +++ b/tests/Fixtures/V2/EntityReferences/OrderRetryActivity.php @@ -0,0 +1,25 @@ +attemptCount() === 1) { + throw new \RuntimeException('Retry after the order changes.'); + } + return $order->id . ':' . $order->status . ':' . $note; + } +} diff --git a/tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php b/tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php index a1ad0cb6..c1316d30 100644 --- a/tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php +++ b/tests/Fixtures/V2/EntityReferences/OrderStatusWorkflow.php @@ -6,14 +6,31 @@ use function Workflow\V2\activity; use Workflow\V2\Attributes\Type; +use function Workflow\V2\localActivity; use Workflow\V2\Workflow; #[Type('tests.entity-references.order-status-workflow')] final class OrderStatusWorkflow extends Workflow { - public function handle(string $orderId): string + public function handle(string $orderId, string $mode = 'queued'): string { - // Workflow code holds the domain entity; the engine records a reference. - return activity(OrderStatusActivity::class, new Order($orderId, 'as-seen-by-workflow'), 'checked'); + $order = new Order($orderId, 'as-seen-by-workflow'); + if ($mode === 'local') { + return localActivity(OrderStatusActivity::class, $order, 'checked'); + } + if ($mode === 'map') { + return activity(OrderMapActivity::class, [ + 'type' => 'orders.order', + 'id' => $orderId, + 'version' => 1, + ], 'checked'); + } + $activity = match ($mode) { + 'injected' => OrderInjectedActivity::class, + 'mismatch' => OrderMismatchActivity::class, + 'retry' => OrderRetryActivity::class, + default => OrderStatusActivity::class, + }; + return activity($activity, $order, 'checked'); } } diff --git a/tests/Fixtures/V2/ReplayRegression/entity-reference-attempt-current-state.json b/tests/Fixtures/V2/ReplayRegression/entity-reference-attempt-current-state.json new file mode 100644 index 00000000..6894c52e --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/entity-reference-attempt-current-state.json @@ -0,0 +1,47 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "entity-reference-attempt-current-state", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "consumers": [ + "workflow-fiber-runner" + ], + "workflow": { + "type": "Tests\\Fixtures\\V2\\EntityReferences\\OrderStatusWorkflow", + "arguments": [ + "corpus-order" + ], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": {}, + "recorded_at": "2026-11-13T10:00:00.000000Z" + } + ], + "expected": { + "completed": false, + "result": null, + "commands": [ + { + "type": "schedule_activity" + } + ] + }, + "entity_reference_attempt": { + "issue": "durable-workflow/workflow#996", + "logical_type": "orders.order", + "entity_id": "corpus-order", + "recorded_positions": [ + 0 + ], + "state_at_attempt": "shipped", + "expected_result": "corpus-order:shipped:checked", + "loads_after_replay": 1 + } +} diff --git a/tests/Unit/V2/EntityReferenceArgumentsTest.php b/tests/Unit/V2/EntityReferenceArgumentsTest.php new file mode 100644 index 00000000..2e90eff2 --- /dev/null +++ b/tests/Unit/V2/EntityReferenceArgumentsTest.php @@ -0,0 +1,159 @@ +previous = Container::getInstance(); + $container = new Container(); + $container->instance('config', new Repository([ + 'workflows' => [ + 'v2' => [ + 'entity_references' => [ + 'orders.order' => [ + 'entity_class' => Order::class, + 'adapter' => OrderReferenceAdapter::class, + 'resolver' => EntityArgumentResolver::class, + ], + ], + ], + ], + ])); + $container->instance(LoggerInterface::class, new NullLogger()); + Container::setInstance($container); + EntityArgumentResolver::$calls = 0; + } + + protected function tearDown(): void + { + Container::setInstance($this->previous); + parent::tearDown(); + } + + public function testReferenceResolutionPreservesAnInjectedServiceAndRecordedPosition(): void + { + $activity = $this->activity([0]); + $method = new ReflectionMethod($activity, 'handle'); + $arguments = EntityReferenceArguments::resolve($activity, $this->arguments(), $method); + $resolved = ActivityEntryArguments::resolve($activity, $arguments, $method); + self::assertInstanceOf(NullLogger::class, $resolved[0]); + self::assertInstanceOf(Order::class, $resolved[1]); + self::assertSame('note', $resolved[2]); + self::assertSame(1, EntityArgumentResolver::$calls); + } + + #[DataProvider('invalidPositions')] + public function testMalformedRecordedPositionsRefuseBeforeResolverAccess(mixed $positions): void + { + $activity = $this->activity($positions); + try { + EntityReferenceArguments::resolve($activity, $this->arguments(), new ReflectionMethod($activity, 'handle')); + self::fail('Malformed metadata was accepted.'); + } catch (EntityReferenceRefused $error) { + self::assertSame(EntityReferenceRefused::INVALID_REFERENCE, $error->reason); + self::assertSame(0, EntityArgumentResolver::$calls); + } + } + + public static function invalidPositions(): iterable + { + yield 'null' => [null]; + yield 'boolean' => [false]; + yield 'string' => ['0']; + yield 'boolean member' => [[false]]; + yield 'negative member' => [[-1]]; + yield 'non-reference member' => [[1]]; + yield 'duplicate member' => [[0, 0]]; + yield 'non-list' => [[ + 1 => 0, + ]]; + yield 'array member' => [[[ + 'position' => 0, + ]]]; + } + + public function testAmbiguousAdaptersRefuseRatherThanChoosingRegistrationOrder(): void + { + $registrations = config('workflows.v2.entity_references'); + $registrations['other.order'] = $registrations['orders.order']; + config() + ->set('workflows.v2.entity_references', $registrations); + $this->expectException(EntityReferenceRefused::class); + EntityReferenceArguments::encode([new Order('id', 'paid')]); + } + + private function activity(mixed $positions): Activity + { + return RuntimeObjectFactory::activity( + EntityArgumentActivity::class, + new ActivityExecution([ + 'id' => 'execution-id', + 'attempt_count' => 1, + 'activity_options' => [ + 'entity_reference_positions' => $positions, + ], + ]), + new WorkflowRun([ + 'id' => 'run-id', + 'namespace' => 'tenant-a', + 'workflow_instance_id' => 'instance-id', + ]), + 'task-id', + ); + } + + private function arguments(): array + { + return [[ + 'type' => 'orders.order', + 'id' => 'id', + 'version' => 1, + ], 'note']; + } +} + +final class EntityArgumentResolver implements EntityReferenceResolver +{ + public static int $calls = 0; + + public function resolve(EntityReference $reference, EntityReferenceContext $context): object + { + ++self::$calls; + return new Order($reference->id, 'paid'); + } +} + +final class EntityArgumentActivity extends Activity +{ + public function handle(LoggerInterface $logger, Order $order, string $note): string + { + return $order->status . ':' . $note; + } +} From 72c5397aa624e5418b57425685038145dfb58b47 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:30:24 +0000 Subject: [PATCH 4/4] Retain the entity reference outcome in replay corpus evidence --- ...ntity-reference-attempt-current-state.json | 41 +++++++++++++++++-- 1 file changed, 38 insertions(+), 3 deletions(-) diff --git a/tests/Fixtures/V2/ReplayRegression/entity-reference-attempt-current-state.json b/tests/Fixtures/V2/ReplayRegression/entity-reference-attempt-current-state.json index 6894c52e..7fd53577 100644 --- a/tests/Fixtures/V2/ReplayRegression/entity-reference-attempt-current-state.json +++ b/tests/Fixtures/V2/ReplayRegression/entity-reference-attempt-current-state.json @@ -22,14 +22,49 @@ "event_type": "WorkflowStarted", "payload": {}, "recorded_at": "2026-11-13T10:00:00.000000Z" + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "recorded_at": "2026-11-13T10:00:00.000001Z", + "payload": { + "sequence": 1, + "activity_execution_id": "01j00000000000000000000e01", + "activity_class": "Tests\\Fixtures\\V2\\EntityReferences\\OrderStatusActivity", + "activity_type": "tests.entity-references.order-status", + "activity": { + "id": "01j00000000000000000000e01", + "sequence": 1, + "class": "Tests\\Fixtures\\V2\\EntityReferences\\OrderStatusActivity", + "type": "tests.entity-references.order-status", + "arguments": "wwHioz3/VYAiNwwEDgYIdHlwZQoYb3JkZXJzLm9yZGVyBGlkChhjb3JwdXMtb3JkZXIOdmVyc2lvbgQCAAoOY2hlY2tlZAA=", + "payload_codec": "avro", + "entity_reference_positions": [ + 0 + ] + } + } + }, + { + "sequence": 3, + "event_type": "ActivityCompleted", + "recorded_at": "2026-11-13T10:00:01.000000Z", + "payload": { + "sequence": 1, + "activity_execution_id": "01j00000000000000000000e01", + "activity_class": "Tests\\Fixtures\\V2\\EntityReferences\\OrderStatusActivity", + "activity_type": "tests.entity-references.order-status", + "result": "wwHioz3/VYAiNwo4Y29ycHVzLW9yZGVyOnNoaXBwZWQ6Y2hlY2tlZA==", + "payload_codec": "avro" + } } ], "expected": { - "completed": false, - "result": null, + "completed": true, + "result": "corpus-order:shipped:checked", "commands": [ { - "type": "schedule_activity" + "type": "complete_workflow" } ] },