diff --git a/CHANGELOG.md b/CHANGELOG.md index cb8e33ef..2d3124bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,13 @@ project follows [Semantic Versioning](https://semver.org/). ### Added +- `ActivityContext::activityExecutionId` exposes the received logical activity + execution ID for downstream idempotency keys. Missing or invalid legacy + metadata stays null; task and attempt IDs are never used as substitutes. + Prepared local callbacks receive their validated Server execution identity; + inline local and Server-free callbacks without one report null + (durable-workflow/.github#211). + - `Client::iterateWorkflows()` and `iterateWorkflowHistory()` lazily read complete workflow listings and exact-run histories, retaining one page at a time. Existing one-page methods stay available. Iteration preserves opaque diff --git a/docs/portal/build/workflows-activities.md b/docs/portal/build/workflows-activities.md index d4498bc5..52fec96c 100644 --- a/docs/portal/build/workflows-activities.md +++ b/docs/portal/build/workflows-activities.md @@ -517,6 +517,30 @@ heartbeats. Prepared local activities use supervised callback processes. ## Inspect retry budget and the owning workflow +The unreleased `ActivityContext::activityExecutionId` property contains the +logical `activity_execution_id` received with the claim. Ordinary and +cooperatively supervised remote callbacks expose that same execution identity +across retry attempts, while `activityAttemptId` identifies each attempt. +Use the logical ID in a downstream idempotency key when the destination honors +that key, including your application scope where needed. + +```php +function activityIdempotencyKey( + \DurableWorkflow\Worker\ActivityContext $context, + string $businessKey, +): string { + return $context->activityExecutionId ?? $businessKey; +} +``` + +Older claims can omit the field. Missing, null, empty or non-string values +produce null, so supply a business key when it is unavailable. Prepared local +callbacks expose the execution ID from their validated Server admission. +Inline local and Server-free callbacks without that admission report null. +Session activity callbacks expose the field when their actual claim includes +it; omission also produces null. The property is read-only and performs no +additional runtime request. + PHP SDK 3.6.0 introduced retry metadata on `ActivityContext`. Use 3.6.1 or later for the original owning workflow and run IDs from public activity claims. `$context->retryInfo` retains the original attempt number and retry policy, diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 988f6367..3391bcaa 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -295,6 +295,18 @@ the supported runtime URL forms, role-specific environment variables, package-owned source files, and expected result. Documentation builds verify that the visible examples still match those runnable files. +## Activity execution identity + +The unreleased `ActivityContext::activityExecutionId` is the nullable logical +execution identity supplied by an activity claim. `workflowId` and `runId` +identify its owning workflow/run; `activityAttemptId` remains attempt-specific. +Use a caller-owned business key when logical identity is unavailable. The SDK +preserves received ID spelling and never manufactures an execution ID from a +task, attempt or session ID. Prepared local callbacks expose the validated +Server execution ID; inline local/Server-free callbacks without one report +null. Session callbacks use the actual claim field when present. The same +accessor is used in ordinary and cooperative remote execution. + ## Workflow handles and control-plane APIs `WorkflowHandle` distinguishes the stable workflow instance from a selected diff --git a/src/Worker.php b/src/Worker.php index e40f067c..9be7f4e0 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -2337,7 +2337,8 @@ function (RequestBudget $budget): void { $this->heartbeatIfDue($budget); }, } return $this->invokeActivity($handler, new ActivityContext($this->client, $attempt->taskId, $attempt->attemptId, $attempt->leaseOwner, $activityType, $attempt->attemptNumber, localHeartbeat: $heartbeat, - taskMetadata: ['workflow_instance_id' => $task['workflow_id'] ?? null, 'workflow_run_id' => $task['run_id'] ?? null]), $arguments, + taskMetadata: ['workflow_instance_id' => $task['workflow_id'] ?? null, 'workflow_run_id' => $task['run_id'] ?? null, + 'activity_execution_id' => $attempt->executionId]), $arguments, ['activity_id' => $attempt->executionId] + ExecutionTelemetry::taskFields($task, $this->client->namespace, $this->taskQueue), true); }; diff --git a/src/Worker/ActivityContext.php b/src/Worker/ActivityContext.php index ca0471b8..b38c56e4 100644 --- a/src/Worker/ActivityContext.php +++ b/src/Worker/ActivityContext.php @@ -16,6 +16,8 @@ final class ActivityContext public readonly ?string $workflowId; /** Original run_id from the public claim; unavailable legacy identity stays null. */ public readonly ?string $runId; + /** Logical execution from the claim, never inferred from task/attempt identity. */ + public readonly ?string $activityExecutionId; /** * @param (Closure(array): array)|null $heartbeatRequest @@ -39,6 +41,7 @@ public function __construct( ? $taskMetadata['workflow_id'] : ($taskMetadata['workflow_instance_id'] ?? null)); $this->runId = self::receivedIdentity(array_key_exists('run_id', $taskMetadata) ? $taskMetadata['run_id'] : ($taskMetadata['workflow_run_id'] ?? null)); + $this->activityExecutionId = self::receivedIdentity($taskMetadata['activity_execution_id'] ?? null); } private static function receivedIdentity(mixed $value): ?string diff --git a/tests/ActivityExecutionIdentityTest.php b/tests/ActivityExecutionIdentityTest.php new file mode 100644 index 00000000..ef3e091b --- /dev/null +++ b/tests/ActivityExecutionIdentityTest.php @@ -0,0 +1,83 @@ +, ?string}> */ + public static function receivedIdentities(): iterable + { + yield 'canonical identity' => [['activity_execution_id' => 'logical-execution'], 'logical-execution']; + yield 'opaque spelling' => [['activity_execution_id' => '01AbC'], '01AbC']; + yield 'zero string' => [['activity_execution_id' => '0'], '0']; + yield 'legacy omission' => [[], null]; + yield 'explicit null' => [['activity_execution_id' => null, 'execution_id' => 'unrelated'], null]; + yield 'empty' => [['activity_execution_id' => ''], null]; + yield 'numeric' => [['activity_execution_id' => 42], null]; + yield 'boolean' => [['activity_execution_id' => true], null]; + yield 'unrelated aliases' => [['execution_id' => 'other', 'task_id' => 'task', 'activity_attempt_id' => 'attempt'], null]; + } + + #[DataProvider('receivedIdentities')] + public function testOnlyTheReceivedLogicalIdentityIsExposed(array $metadata, ?string $expected): void + { + $context = new ActivityContext(new Client('https://server.example'), 'task', 'attempt', 'owner', 'activity', 1, + taskMetadata: $metadata); + self::assertSame($expected, $context->activityExecutionId); + self::assertSame('task', $context->taskId); + self::assertSame('attempt', $context->activityAttemptId); + } + + public function testOldConstructorsRemainValidWithoutInventingIdentity(): void + { + $context = new ActivityContext(new Client('https://server.example'), 'task', 'attempt', 'owner', 'activity', 1); + self::assertNull($context->activityExecutionId); + } + + public function testReceivedRetryClaimsKeepLogicalIdentityAndDifferentAttempts(): void + { + $polls = 0; + $failures = 0; + $completions = 0; + $seen = []; + $transport = new FakeTransport(handler: static function (string $method, string $uri) use (&$polls, &$failures, &$completions): array { + if (str_ends_with($uri, '/worker/register')) { return ['registered' => true]; } + if (str_ends_with($uri, '/activity-tasks/poll')) { + ++$polls; + return ['task' => ['task_id' => 'task-'.$polls, 'activity_attempt_id' => 'attempt-'.$polls, + 'activity_execution_id' => 'logical-execution', 'workflow_id' => 'original-workflow', 'run_id' => 'original-run', + 'lease_owner' => 'owner', 'activity_type' => 'identity', 'attempt_number' => $polls, 'payload_codec' => 'avro']]; + } + if (str_ends_with($uri, '/fail')) { ++$failures; return ['failed' => true]; } + if (str_ends_with($uri, '/complete')) { ++$completions; return ['completed' => true]; } + if (str_ends_with($uri, '/poll')) { return ['task' => null]; } + self::fail('Unexpected request: '.$method.' '.$uri); + }); + $worker = new Worker(new Client('https://server.example', transport: $transport), 'queue', workerId: 'owner'); + $worker->registerActivity('identity', static function (ActivityContext $context) use (&$seen): string { + $seen[] = [$context->activityExecutionId, $context->activityAttemptId, $context->workflowId, $context->runId]; + if ($context->attemptNumber === 1) { + throw new RuntimeException('Retry the received activity.'); + } + return 'done'; + }); + self::assertTrue($worker->tick(0)); + self::assertTrue($worker->tick(0)); + self::assertSame([ + ['logical-execution', 'attempt-1', 'original-workflow', 'original-run'], + ['logical-execution', 'attempt-2', 'original-workflow', 'original-run'], + ], $seen); + self::assertSame(1, $failures); + self::assertSame(1, $completions); + } +} diff --git a/tests/ActivityRetryInfoTest.php b/tests/ActivityRetryInfoTest.php index 1fa67ac3..4d96e90b 100644 --- a/tests/ActivityRetryInfoTest.php +++ b/tests/ActivityRetryInfoTest.php @@ -67,6 +67,7 @@ public function testLegacyConstructorAndServerFreeActivitiesHaveUnknownMetadata( self::assertNull($context->retryInfo->isLastConfiguredAttempt()); self::assertNull($context->workflowId); self::assertNull($context->runId); + self::assertNull($context->activityExecutionId); } public static function originalIdentities(): array @@ -123,13 +124,15 @@ public function testInlineLocalActivityRetainsParentIdentityWithoutInventingRetr $worker->registerWorkflow('flow', static fn (WorkflowContext $context): mixed => $context->localActivity('metadata')); $worker->registerActivity('metadata', static fn (ActivityContext $context): array => [ 'workflow_id' => $context->workflowId, 'run_id' => $context->runId, + 'activity_execution_id' => $context->activityExecutionId, 'provided' => $context->retryInfo->policyProvided, 'last' => $context->retryInfo->isLastConfiguredAttempt(), ]); self::assertTrue($worker->tick(0)); self::assertNotNull($completion); $finished = array_values(array_filter($completion['commands'], static fn (array $command): bool => $command['type'] === 'complete_workflow')); self::assertCount(1, $finished); - self::assertSame(['workflow_id' => 'original-workflow', 'run_id' => 'original-run', 'provided' => false, 'last' => null], + self::assertSame(['workflow_id' => 'original-workflow', 'run_id' => 'original-run', 'activity_execution_id' => null, + 'provided' => false, 'last' => null], (new AvroPayloadCodec())->decodeEnvelope($finished[0]['result'])); } diff --git a/tests/PreparedLocalActivityWorkerTest.php b/tests/PreparedLocalActivityWorkerTest.php index 5f09e1dd..65fec2b6 100644 --- a/tests/PreparedLocalActivityWorkerTest.php +++ b/tests/PreparedLocalActivityWorkerTest.php @@ -47,6 +47,7 @@ public function test_prepared_activity_retains_parent_ids_with_unknown_server_re $worker->registerWorkflow('prepared', static fn (WorkflowContext $context): mixed => $context->localActivity('effect')); $worker->registerActivity('effect', static fn (ActivityContext $context): array => [ 'workflow_id' => $context->workflowId, 'run_id' => $context->runId, + 'activity_execution_id' => $context->activityExecutionId, 'provided' => $context->retryInfo->policyProvided, 'last' => $context->retryInfo->isLastConfiguredAttempt(), ]); $worker->run(0); @@ -54,7 +55,8 @@ public function test_prepared_activity_retains_parent_ids_with_unknown_server_re self::assertCount(1, $transport->operations('outcome')); $command = $transport->completions[0]['commands'][0]; self::assertSame('complete_workflow', $command['type']); - self::assertSame(['workflow_id' => 'workflow', 'run_id' => 'run', 'provided' => false, 'last' => null], + self::assertSame(['workflow_id' => 'workflow', 'run_id' => 'run', 'activity_execution_id' => 'backend-execution-1', + 'provided' => false, 'last' => null], (new AvroPayloadCodec())->decodeEnvelope($command['result'])); }