Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
24 changes: 24 additions & 0 deletions docs/portal/build/workflows-activities.md
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
12 changes: 12 additions & 0 deletions docs/sdk-reference.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 2 additions & 1 deletion src/Worker.php
Original file line number Diff line number Diff line change
Expand Up @@ -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);
};

Expand Down
3 changes: 3 additions & 0 deletions src/Worker/ActivityContext.php
Original file line number Diff line number Diff line change
Expand Up @@ -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-key, mixed>): array<string, mixed>)|null $heartbeatRequest
Expand All @@ -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
Expand Down
83 changes: 83 additions & 0 deletions tests/ActivityExecutionIdentityTest.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
<?php

declare(strict_types=1);

namespace DurableWorkflow\Tests;

use DurableWorkflow\Client;
use DurableWorkflow\Tests\Support\FakeTransport;
use DurableWorkflow\Worker;
use DurableWorkflow\Worker\ActivityContext;
use PHPUnit\Framework\Attributes\DataProvider;
use PHPUnit\Framework\TestCase;
use RuntimeException;

final class ActivityExecutionIdentityTest extends TestCase
{
/** @return iterable<string, array{array<string, mixed>, ?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);
}
}
5 changes: 4 additions & 1 deletion tests/ActivityRetryInfoTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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']));
}

Expand Down
4 changes: 3 additions & 1 deletion tests/PreparedLocalActivityWorkerTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -47,14 +47,16 @@ 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);
self::assertSame([], $transport->failures);
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']));
}

Expand Down
Loading