From b4c165895c30af50d993016bde017567d1ce2e6d Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 03:58:49 +0000 Subject: [PATCH 1/6] Specify bounded activity lease renewal without user heartbeats --- tests/ActivityLeaseRenewalClientTest.php | 200 +++++++++++++++++++++++ 1 file changed, 200 insertions(+) create mode 100644 tests/ActivityLeaseRenewalClientTest.php diff --git a/tests/ActivityLeaseRenewalClientTest.php b/tests/ActivityLeaseRenewalClientTest.php new file mode 100644 index 0000000..72c46b6 --- /dev/null +++ b/tests/ActivityLeaseRenewalClientTest.php @@ -0,0 +1,200 @@ + true, 'heartbeat_recorded' => false, 'original' => 'retained']; + $http = new GuzzleClient(['handler' => static function (RequestInterface $request, array $options) use (&$requests, $ack) { + $requests[] = ['request' => $request, 'timeout' => $options['timeout']]; + + return Create::promiseFor(new Response(200, [], json_encode( + count($requests) === 1 ? self::discovery(true) : $ack, JSON_THROW_ON_ERROR, + ))); + }]); + $client = new Client('https://server.example', transport: new Psr18Transport($http), + namespace: 'tenant-a', controlToken: 'control', workerToken: 'worker'); + + self::assertSame($ack, $client->renewActivityTaskLease('task/0', 'attempt/17', 'owner')); + self::assertCount(2, $requests); + self::assertSame('https://server.example/api/cluster/info', (string) $requests[0]['request']->getUri()); + self::assertSame('https://server.example/api/worker/activity-tasks/task%2F0/renew-lease', (string) $requests[1]['request']->getUri()); + self::assertSame(['activity_attempt_id' => 'attempt/17', 'lease_owner' => 'owner'], + json_decode((string) $requests[1]['request']->getBody(), true, 512, JSON_THROW_ON_ERROR)); + foreach ($requests as $call) { + self::assertSame('Bearer worker', $call['request']->getHeaderLine('Authorization')); + self::assertSame('tenant-a', $call['request']->getHeaderLine('X-Namespace')); + self::assertSame('1.20', $call['request']->getHeaderLine('X-Durable-Workflow-Protocol-Version')); + self::assertFalse($call['request']->hasHeader('X-Durable-Workflow-Control-Plane-Version')); + self::assertGreaterThan(0, $call['timeout']); + self::assertLessThanOrEqual(5, $call['timeout']); + } + } + + #[DataProvider('incapableRuntimeProvider')] + public function testOnlyAffirmativeCapabilityAllowsRenewal(mixed $capability): void + { + $calls = 0; + $http = new GuzzleClient(['handler' => static function () use (&$calls, $capability) { + ++$calls; + + return Create::promiseFor(new Response(200, [], json_encode(self::discovery($capability), JSON_THROW_ON_ERROR))); + }]); + $client = new Client('https://server.example', transport: new Psr18Transport($http)); + try { + $client->renewActivityTaskLease('task', 'attempt', 'owner'); + self::fail('An unavailable or malformed capability allowed activity renewal.'); + } catch (LogicException $error) { + self::assertStringContainsString('activity_lease_renewal', $error->getMessage()); + } + self::assertSame(1, $calls, 'Capability refusal must not submit a renewal.'); + } + + public static function incapableRuntimeProvider(): array + { + return [[false], [null], ['true'], [1], [[]]]; + } + + public function testExistingProtocolWithAffirmativeCapabilityAddsNoNewProtocolFloor(): void + { + $http = new GuzzleClient(['handler' => static fn (): mixed => Create::promiseFor( + new Response(200, [], json_encode([...self::discovery(true), 'lease_renewed' => true], JSON_THROW_ON_ERROR)), + )]); + $client = new Client('https://server.example', transport: new Psr18Transport($http), workerProtocolVersion: '1.19'); + + self::assertTrue($client->renewActivityTaskLease('0', '017', '17')['lease_renewed']); + } + + #[DataProvider('invalidIdentityProvider')] + public function testInvalidIdentityRefusesBeforeDiscovery(string $task, string $attempt, string $owner): void + { + $transport = new FakeTransport(); + $client = new Client('https://server.example', transport: $transport); + try { + $client->renewActivityTaskLease($task, $attempt, $owner); + self::fail('An invalid claim allowed discovery.'); + } catch (InvalidArgumentException) { + self::assertSame([], $transport->requests); + } + } + + public static function invalidIdentityProvider(): array + { + return [['', 'attempt', 'owner'], ['task', ' ', 'owner'], ['task', 'attempt', ''], + [str_repeat('x', 256), 'attempt', 'owner'], ['task', str_repeat('x', 256), 'owner'], + ['task', 'attempt', str_repeat('x', 256)]]; + } + + public function testUnboundedTransportCannotClaimToMaintainALease(): void + { + $transport = new FakeTransport(); + $client = new Client('https://server.example', transport: $transport); + try { + $client->renewActivityTaskLease('task', 'attempt', 'owner'); + self::fail('A transport without bounded I/O allowed renewal.'); + } catch (InvalidArgumentException $error) { + self::assertStringContainsString('bounded', $error->getMessage()); + } + self::assertSame([], $transport->requests); + } + + public function testExpiredOriginalBudgetCannotDiscoverOrRenew(): void + { + $transport = new FakeTransport(); + $client = new Client('https://server.example', transport: $transport); + $budget = new RequestBudget(5, hrtime(true) / 1e9 - 1); + try { + $client->renewActivityTaskLease('task', 'attempt', 'owner', $budget); + self::fail('An expired original authority allowed I/O.'); + } catch (TransportException $error) { + self::assertFalse($error->transientConnectionFailure); + } + self::assertSame([], $transport->requests); + } + + public function testWideCallerBudgetIsRestrictedToFiveSeconds(): void + { + $timeouts = []; + $http = new GuzzleClient(['handler' => static function (RequestInterface $request, array $options) use (&$timeouts) { + $timeouts[] = $options['timeout']; + + return Create::promiseFor(new Response(200, [], json_encode(self::discovery(true), JSON_THROW_ON_ERROR))); + }]); + $client = new Client('https://server.example', transport: new Psr18Transport($http)); + $client->renewActivityTaskLease('task', 'attempt', 'owner', new RequestBudget(65)); + + self::assertCount(2, $timeouts); + foreach ($timeouts as $timeout) { + self::assertGreaterThan(0, $timeout); + self::assertLessThanOrEqual(5, $timeout); + } + } + + public function testDiscoveryCannotGetAFreshRequestBudgetAfterConsumingTheOriginalWindow(): void + { + $calls = 0; + $http = new GuzzleClient(['handler' => static function () use (&$calls) { + ++$calls; + usleep(1100000); + + return Create::promiseFor(new Response(200, [], json_encode(self::discovery(true), JSON_THROW_ON_ERROR))); + }]); + $client = new Client('https://server.example', transport: new Psr18Transport($http)); + $budget = new RequestBudget(1); + try { + $client->renewActivityTaskLease('task', 'attempt', 'owner', $budget); + self::fail('Discovery refreshed the expired caller window.'); + } catch (ServerException $error) { + self::assertStringContainsString('request time budget expired', $error->getMessage()); + } + self::assertSame(1, $calls); + } + + public function testSuccessfulResponseAfterTheOriginalWindowDoesNotReturnAGrant(): void + { + $calls = 0; + $http = new GuzzleClient(['handler' => static function () use (&$calls) { + if (++$calls === 2) { + usleep(1100000); + } + + return Create::promiseFor(new Response(200, [], json_encode( + $calls === 1 ? self::discovery(true) : ['lease_renewed' => true], JSON_THROW_ON_ERROR, + ))); + }]); + $client = new Client('https://server.example', transport: new Psr18Transport($http)); + $budget = new RequestBudget(1); + try { + $client->renewActivityTaskLease('task', 'attempt', 'owner', $budget); + self::fail('A late successful response acquired fresh execution authority.'); + } catch (ServerException $error) { + self::assertStringContainsString('request time budget expired', $error->getMessage()); + } + self::assertSame(2, $calls); + } + + private static function discovery(mixed $capability): array + { + return ['worker_protocol' => ['server_capabilities' => ['activity_lease_renewal' => $capability]]]; + } +} From 7ab48ce24da7887c206bd1661ce8db3ee3802262 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 04:05:12 +0000 Subject: [PATCH 2/6] Renew original activity runtime leases within one bounded request window --- src/Client.php | 35 ++++++++++++++++++++++++ src/Transport/RequestBudget.php | 9 ++++++ tests/ActivityLeaseRenewalClientTest.php | 31 ++++++++++++++++++--- 3 files changed, 71 insertions(+), 4 deletions(-) diff --git a/src/Client.php b/src/Client.php index d972b07..a396993 100644 --- a/src/Client.php +++ b/src/Client.php @@ -2169,6 +2169,41 @@ public function heartbeatActivityTask( ]); } + /** + * Renew only the original runtime lease, without recording application progress. + * Discovery and transport share at most five seconds and any earlier caller ceiling. + * The caller must validate the returned original identity and actual grant before use. + * + * @return array + */ + public function renewActivityTaskLease( + string $taskId, + string $activityAttemptId, + string $leaseOwner, + ?RequestBudget $budget = null, + ): array { + foreach ([$taskId, $activityAttemptId, $leaseOwner] as $identity) { + if (trim($identity) === '' || strlen($identity) > 255) { + throw new InvalidArgumentException('Activity lease renewal requires bounded, nonempty original claim identities.'); + } + } + $budget ??= new RequestBudget(5); + $budget->restrictRequestDuration(5); + $budget->remainingSeconds(); + $client = $this->withBoundedWorkerRequests(); + $info = $client->worker('GET', '/cluster/info', budget: $budget); + $protocol = $info['worker_protocol'] ?? null; + $capabilities = is_array($protocol) ? ($protocol['server_capabilities'] ?? null) : null; + if (!is_array($capabilities) || ($capabilities['activity_lease_renewal'] ?? null) !== true) { + throw new \LogicException('Runtime must explicitly discover activity_lease_renewal before activity lease maintenance.'); + } + + return $client->worker('POST', '/worker/activity-tasks/'.$this->segment($taskId).'/renew-lease', [ + 'activity_attempt_id' => $activityAttemptId, + 'lease_owner' => $leaseOwner, + ], $budget); + } + /** Observe ownership without renewing a lease or recording user progress. * @return array */ diff --git a/src/Transport/RequestBudget.php b/src/Transport/RequestBudget.php index 85d3e0b..d457911 100644 --- a/src/Transport/RequestBudget.php +++ b/src/Transport/RequestBudget.php @@ -37,6 +37,15 @@ public function restrictAuthorityDeadline(float $deadline): void $this->deadline = min($this->deadline, $this->authorityDeadline); } + /** Cap a sub-operation without replacing the caller's original I/O or authority clock. */ + public function restrictRequestDuration(int $seconds): void + { + if ($seconds < 1 || $seconds > 65) { + throw new \InvalidArgumentException('A worker request budget must be from 1 through 65 seconds.'); + } + $this->deadline = min($this->deadline, hrtime(true) / 1e9 + $seconds); + } + /** Use the original clock mapping, including for authority learned during I/O. */ public function restrictWallAuthorityDeadline(DateTimeImmutable $deadline): void { diff --git a/tests/ActivityLeaseRenewalClientTest.php b/tests/ActivityLeaseRenewalClientTest.php index 72c46b6..059b6df 100644 --- a/tests/ActivityLeaseRenewalClientTest.php +++ b/tests/ActivityLeaseRenewalClientTest.php @@ -52,13 +52,13 @@ public function testRenewalUsesOnlyWorkerCredentialsAndOriginalClaimWithoutUserP } #[DataProvider('incapableRuntimeProvider')] - public function testOnlyAffirmativeCapabilityAllowsRenewal(mixed $capability): void + public function testOnlyAffirmativeCapabilityAllowsRenewal(array $discovery): void { $calls = 0; - $http = new GuzzleClient(['handler' => static function () use (&$calls, $capability) { + $http = new GuzzleClient(['handler' => static function () use (&$calls, $discovery) { ++$calls; - return Create::promiseFor(new Response(200, [], json_encode(self::discovery($capability), JSON_THROW_ON_ERROR))); + return Create::promiseFor(new Response(200, [], json_encode($discovery, JSON_THROW_ON_ERROR))); }]); $client = new Client('https://server.example', transport: new Psr18Transport($http)); try { @@ -72,7 +72,9 @@ public function testOnlyAffirmativeCapabilityAllowsRenewal(mixed $capability): v public static function incapableRuntimeProvider(): array { - return [[false], [null], ['true'], [1], [[]]]; + return [[self::discovery(false)], [self::discovery(null)], [self::discovery('true')], + [self::discovery(1)], [self::discovery([])], [[]], [['worker_protocol' => 'malformed']], + [['worker_protocol' => ['server_capabilities' => true]]]]; } public function testExistingProtocolWithAffirmativeCapabilityAddsNoNewProtocolFloor(): void @@ -193,6 +195,27 @@ public function testSuccessfulResponseAfterTheOriginalWindowDoesNotReturnAGrant( self::assertSame(2, $calls); } + public function testExpiredLeaseRefusalIsRetainedWithoutRetryingAnotherClaim(): void + { + $calls = 0; + $http = new GuzzleClient(['handler' => static function () use (&$calls) { + ++$calls; + + return Create::promiseFor($calls === 1 + ? new Response(200, [], json_encode(self::discovery(true), JSON_THROW_ON_ERROR)) + : new Response(409, [], '{"reason":"lease_expired","message":"Original lease expired"}')); + }]); + $client = new Client('https://server.example', transport: new Psr18Transport($http)); + try { + $client->renewActivityTaskLease('task', 'attempt', 'owner'); + self::fail('An expired lease acquired execution authority.'); + } catch (ServerException $error) { + self::assertSame(409, $error->status); + self::assertSame('lease_expired', $error->reason); + } + self::assertSame(2, $calls); + } + private static function discovery(mixed $capability): array { return ['worker_protocol' => ['server_capabilities' => ['activity_lease_renewal' => $capability]]]; From dd9748cbc55911954bc6295e79d3ef2cf1506fe3 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 10:25:48 +0000 Subject: [PATCH 3/6] Maintain original PHP activity leases in an independent process --- CHANGELOG.md | 15 + README.md | 18 ++ docs/sdk-reference.md | 52 +++ src/Client.php | 92 +++++- src/Transport/Psr18Transport.php | 8 + src/Worker.php | 77 ++++- src/Worker/ActivityLeaseAuthority.php | 319 +++++++++++++++++++ src/Worker/ActivityLeaseBinding.php | 50 +++ src/Worker/ActivityLeaseMaintainer.php | 249 +++++++++++++++ src/Worker/ActivityLeaseProcess.php | 228 +++++++++++++ src/Worker/ActivityLeaseSession.php | 108 +++++++ src/Worker/activity-lease-maintainer.php | 10 + tests/ActivityLeaseAuthorityTest.php | 246 ++++++++++++++ tests/Support/activity-lease-http.php | 99 ++++++ tests/Support/activity-lease-owner.php | 13 + tests/WorkerActivityLeaseMaintenanceTest.php | 277 ++++++++++++++++ 16 files changed, 1845 insertions(+), 16 deletions(-) create mode 100644 src/Worker/ActivityLeaseAuthority.php create mode 100644 src/Worker/ActivityLeaseBinding.php create mode 100644 src/Worker/ActivityLeaseMaintainer.php create mode 100644 src/Worker/ActivityLeaseProcess.php create mode 100644 src/Worker/ActivityLeaseSession.php create mode 100644 src/Worker/activity-lease-maintainer.php create mode 100644 tests/ActivityLeaseAuthorityTest.php create mode 100644 tests/Support/activity-lease-http.php create mode 100644 tests/Support/activity-lease-owner.php create mode 100644 tests/WorkerActivityLeaseMaintenanceTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 2cb48e0..9f1a646 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,21 @@ project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- Optional, bounded metadata-only activity lease renewal against runtimes that + explicitly advertise it, without a new worker protocol floor. +- Automatic lease and Worker-presence maintenance in an independent PHP CLI + helper while ordinary activity callbacks retain their inline execution and + application-owned service identities. Custom HTTP/authentication setups can + provide an explicit fresh-process Client bootstrap. + +### Fixed + +- Fence late activity outcomes after original lease, business, progress or + required session authority is lost. Maintenance cannot fabricate application + heartbeats, restore an expired attempt or reset an original session TTL. + ## [3.1.0] - 2026-10-21 ### Added diff --git a/README.md b/README.md index b009b03..a95c3a4 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,24 @@ worker configuration, Message Streams, framework setup, and testing examples. The generated [API reference](https://php.durable-workflow.com/api/) documents every public class and method. +## Activity runtime leases + +When the runtime advertises `activity_lease_renewal`, the managed Worker keeps +an activity's runtime lease and Worker presence alive independently of a blocking +callback. Ordinary callbacks keep running inline with their original injected +services and captured memory. This requires PHP CLI with `proc_open` enabled. + +Runtime maintenance does not record application progress or extend an activity, +heartbeat, workflow or worker-session TTL deadline. Continue calling +`ActivityContext::heartbeat()` when an activity has a heartbeat timeout. +After authority is lost, the Worker discards late outcomes. It cannot undo +external side effects or forcibly stop ordinary inline application code. + +The default HTTP transport and static worker credentials work automatically. +Custom HTTP clients or dynamic authentication require +[`withActivityLeaseBootstrap()`](docs/sdk-reference.md#activity-runtime-lease-maintenance) +so the independent process recreates their actual TLS and authentication settings. + ## Cooperative cancellation release candidate Cooperative requests let workflow code perform bounded, replayable cleanup. diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 8c45a75..e3ee358 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -546,6 +546,58 @@ task kind, consecutive attempt, selected delay, and typed server exception. Authentication failures, malformed responses, and generic server errors remain fatal. +## Activity runtime lease maintenance + +The managed Worker starts a separate PHP CLI maintenance process when the +runtime registration advertises `activity_lease_renewal`. It renews only the +original activity attempt and reports the Worker slots as busy. Ordinary +activity code still runs in the original Worker process, including captured +memory, dependency-injected services and application-owned connections. + +Maintenance carries the original namespace, queue, task, run, execution, +attempt and owner. It cannot extend recorded activity/workflow deadlines, +change an attempt or reset the original TTL of a required worker session. +Actual application heartbeats can advance only their recorded progress +interval. Runtime pings do not record progress or add heartbeat history events. +An activity with a heartbeat timeout must still call `$context->heartbeat()`. + +The helper exits when its owning process dies or its private input closes. +Server recovery then uses the last acknowledged lease and recorded retry +policy. Ordinary inline PHP callbacks cannot be forcibly stopped safely by +this helper. If authority is lost, the managed Worker and its captured Client +reject late completion/failure publication even when application code catches +an earlier refusal. External side effects still need application idempotency. + +PHP CLI must permit `proc_open`. The default HTTP transport and static worker +credentials need no extra setup. Custom PSR-18 clients, middleware, TLS/CA +configuration or dynamic authentication require a trusted bootstrap file that +returns a fresh Client for the same runtime URI, namespace and protocol: + +```php +// lease-client.php: recreate the same HTTP/auth settings used by the Worker. +return new DurableWorkflow\Client( + getenv('DURABLE_WORKFLOW_SERVER'), + namespace: 'orders', + workerToken: getenv('DURABLE_WORKFLOW_WORKER_TOKEN'), + transport: new DurableWorkflow\Transport\Psr18Transport( + new GuzzleHttp\Client(['verify' => getenv('COMPANY_CA_BUNDLE')]), + ), +); +``` + +Configure the original Client before constructing the Worker: + +```php +$client = $client->withActivityLeaseBootstrap(__DIR__.'/lease-client.php'); +``` + +The file runs in the helper, so create new transport/authentication objects +there. Do not return inherited database or HTTP handles. Worker credentials +and claim metadata travel over the private input pipe, not command arguments. +The SDK refuses custom configurations without this bootstrap rather than +silently replacing their TLS, proxy or authentication behavior. Runtimes that +do not advertise the capability retain their existing lease behavior. + ## Receive repeated input with Message Streams Inbound Message Streams deliver repeated, ordered application input to a stable diff --git a/src/Client.php b/src/Client.php index a396993..54ade9a 100644 --- a/src/Client.php +++ b/src/Client.php @@ -56,6 +56,8 @@ use DurableWorkflow\Worker\CancellationScopeDeliveryReceipt; use DurableWorkflow\Worker\WorkflowClaimAborted; use DurableWorkflow\Worker\WorkerSessionOptions; +use DurableWorkflow\Worker\ActivityLeaseBinding; +use DurableWorkflow\Worker\ActivityLeaseProcess; use InvalidArgumentException; /** Synchronous control-plane and worker-plane client for the standalone server. */ @@ -64,6 +66,7 @@ final class Client implements WorkflowClientInterface private const EXTERNAL_PAYLOAD_REFERENCE_SCHEMA = 'durable-workflow.v2.external-payload-reference.v1'; private const WORKFLOW_TASK_WAITING_FOR_HISTORY_MESSAGE = 'Workflow task waiting for scheduled history.'; private const WORKFLOW_TASK_WAITING_FOR_HISTORY_TYPE = 'WorkflowTaskWaitingForHistory'; + private const ACTIVITY_TASK_PATH = '/worker/activity-tasks/'; private readonly string $baseUri; private readonly ?Authentication $authentication; @@ -71,6 +74,8 @@ final class Client implements WorkflowClientInterface private readonly PayloadCodec $codec; private readonly RuntimePayloadUploads $payloadUploads; private bool $boundedWorkerRequests = false; + private ActivityLeaseBinding $activityLeaseBinding; + private ?string $activityLeaseBootstrap = null; private ?ExecutionTelemetry $executionTelemetry = null; private bool $observabilityContextSupported = false; /** @var (\Closure(string, array|null, array, RequestBudget|null): (?\Closure))|null */ @@ -124,6 +129,7 @@ public function __construct( $this->transport = $transport ?? new Psr18Transport(); $this->codec = $codec ?? new AvroPayloadCodec(); $this->payloadUploads = new RuntimePayloadUploads($this->transport, $this->baseUri); + $this->activityLeaseBinding = new ActivityLeaseBinding(); } public function payloadCodec(): PayloadCodec @@ -131,6 +137,54 @@ public function payloadCodec(): PayloadCodec return $this->codec; } + /** + * Recreate custom authentication and HTTP configuration in the independent + * lease-maintenance process. The trusted PHP file must return a Client with + * this same runtime URI, namespace and protocol. It never runs user activity code. + */ + public function withActivityLeaseBootstrap(string $path): self + { + $resolved = realpath($path); + if ($resolved === false || !is_file($resolved) || !is_readable($resolved)) { + throw new InvalidArgumentException('Activity lease bootstrap must be a readable PHP file.'); + } + $copy = clone $this; + $copy->activityLeaseBootstrap = $resolved; + + return $copy; + } + + /** @internal Credentials travel only over the private helper input pipe. + * @return array + */ + public function activityLeaseProcessConfiguration(): array + { + $configuration = ['base_uri' => $this->baseUri, 'namespace' => $this->namespace, + 'worker_protocol' => $this->workerProtocolVersion, 'bootstrap' => $this->activityLeaseBootstrap]; + if ($this->activityLeaseBootstrap !== null) { + return $configuration; + } + if (!$this->transport instanceof Psr18Transport || !$this->transport->supportsDefaultActivityLeaseProcess() + || ($this->authentication !== null && !$this->authentication instanceof TokenAuthentication + && !$this->authentication instanceof \DurableWorkflow\Auth\StaticHeadersAuthentication)) { + throw new InvalidArgumentException('Independent activity maintenance requires withActivityLeaseBootstrap() for custom authentication or HTTP clients; their TLS and transport settings cannot be discarded.'); + } + + return $configuration + ['worker_headers' => $this->authentication?->headers(true) ?? []]; + } + + /** @internal Bootstrap validation never makes a control-plane request. */ + public function activityLeaseEndpoint(): string + { + return $this->baseUri; + } + + /** @internal Bind only the current execution context, including cloned Clients. */ + public function bindActivityLease(ActivityLeaseProcess $process): \Closure + { + return $this->activityLeaseBinding->enter($process); + } + /** Configure tracing before explicit cluster discovery and business I/O. */ public function withTelemetry(ExecutionTelemetry $telemetry): self { @@ -193,14 +247,21 @@ public function supportsSupervisedQueryRequests(): bool /** Return a new client with the same transport, authentication, and codec for another namespace. */ public function withNamespace(string $namespace): self { - return new self( + $copy = new self( $this->baseUri, $this->authentication, $namespace, $this->transport, $this->codec, maxExternalPayloadBytes: $this->maxExternalPayloadBytes, + workerProtocolVersion: $this->workerProtocolVersion, ); + $copy->activityLeaseBootstrap = $this->activityLeaseBootstrap; + if ($namespace === $this->namespace) { + $copy->activityLeaseBinding = $this->activityLeaseBinding; + } + + return $copy; } /** @return array */ @@ -2120,7 +2181,7 @@ public function completeActivityTask( string $leaseOwner, mixed $result, ): array { - return $this->worker('POST', '/worker/activity-tasks/'.$this->segment($taskId).'/complete', [ + return $this->worker('POST', self::ACTIVITY_TASK_PATH.$this->segment($taskId).'/complete', [ 'activity_attempt_id' => $activityAttemptId, 'lease_owner' => $leaseOwner, 'result' => $this->codec->envelope($result), @@ -2140,7 +2201,7 @@ public function failActivityTask( bool $nonRetryable = false, ?array $details = null, ): array { - return $this->worker('POST', '/worker/activity-tasks/'.$this->segment($taskId).'/fail', [ + return $this->worker('POST', self::ACTIVITY_TASK_PATH.$this->segment($taskId).'/fail', [ 'activity_attempt_id' => $activityAttemptId, 'lease_owner' => $leaseOwner, 'failure' => $this->withoutNulls([ @@ -2162,7 +2223,7 @@ public function heartbeatActivityTask( string $leaseOwner, array $details = [], ): array { - return $this->worker('POST', '/worker/activity-tasks/'.$this->segment($taskId).'/heartbeat', [ + return $this->worker('POST', self::ACTIVITY_TASK_PATH.$this->segment($taskId).'/heartbeat', [ 'activity_attempt_id' => $activityAttemptId, 'lease_owner' => $leaseOwner, 'details' => $details, @@ -2198,7 +2259,7 @@ public function renewActivityTaskLease( throw new \LogicException('Runtime must explicitly discover activity_lease_renewal before activity lease maintenance.'); } - return $client->worker('POST', '/worker/activity-tasks/'.$this->segment($taskId).'/renew-lease', [ + return $client->worker('POST', self::ACTIVITY_TASK_PATH.$this->segment($taskId).'/renew-lease', [ 'activity_attempt_id' => $activityAttemptId, 'lease_owner' => $leaseOwner, ], $budget); @@ -2213,7 +2274,7 @@ public function activityTaskStatus(string $taskId, string $activityAttemptId, st throw new InvalidArgumentException('Activity attempt observation requires worker protocol 1.20.'); } - return $this->worker('POST', '/worker/activity-tasks/'.$this->segment($taskId).'/status', [ + return $this->worker('POST', self::ACTIVITY_TASK_PATH.$this->segment($taskId).'/status', [ 'activity_attempt_id' => $activityAttemptId, 'lease_owner' => $leaseOwner, ], $budget); @@ -2238,7 +2299,7 @@ public function acknowledgeActivityCancellation( } } - return $this->worker('POST', '/worker/activity-tasks/'.$this->segment($taskId).'/acknowledge-cancellation', [ + return $this->worker('POST', self::ACTIVITY_TASK_PATH.$this->segment($taskId).'/acknowledge-cancellation', [ 'activity_attempt_id' => $activityAttemptId, 'lease_owner' => $leaseOwner, 'request_id' => $requestId, @@ -2334,7 +2395,13 @@ private function control( */ private function worker(string $method, string $path, ?array $body = null, ?RequestBudget $budget = null): array { - if ($budget !== null && !$this->boundedWorkerRequests) { + $lease = $this->activityLeaseBinding->current(); + $governed = $lease !== null && $lease->governs($path); + if ($governed) { + $lease->checkRequest($path, $body); + $budget ??= $lease->budget(); + } + if ($budget !== null && !$this->boundedWorkerRequests && !$governed) { throw new \LogicException('An activity receipt budget requires bounded worker requests.'); } if ($method === 'POST' @@ -2347,7 +2414,14 @@ private function worker(string $method, string $path, ?array $body = null, ?Requ $this->assertWorkflowCommandPayloads($commands); } - return $this->request($method, $path, true, $body, budget: $budget); + $started = hrtime(true) / 1e9; + $requestClient = $governed ? $this->withBoundedWorkerRequests() : $this; + $response = $requestClient->request($method, $path, true, $body, budget: $budget); + if ($governed && str_ends_with($path, '/heartbeat')) { + $lease->progress($response, $started, hrtime(true) / 1e9); + } + + return $response; } /** @param list $commands */ diff --git a/src/Transport/Psr18Transport.php b/src/Transport/Psr18Transport.php index df1e10c..8b2a835 100644 --- a/src/Transport/Psr18Transport.php +++ b/src/Transport/Psr18Transport.php @@ -27,6 +27,7 @@ final class Psr18Transport implements BoundedPayloadTransport, BoundedPayloadUpl private readonly ClientInterface $client; private readonly RequestFactoryInterface $requestFactory; private readonly StreamFactoryInterface $streamFactory; + private readonly bool $defaultProcessConfiguration; public function __construct( ?ClientInterface $client = null, @@ -38,6 +39,7 @@ public function __construct( throw new \InvalidArgumentException('Supervised payload transfers require a timeout from 1 through 900 seconds.'); } $factory = new HttpFactory(); + $this->defaultProcessConfiguration = $client === null && $requestFactory === null && $streamFactory === null; $this->client = $client ?? new GuzzleClient(['http_errors' => false]); $this->requestFactory = $requestFactory ?? $factory; $this->streamFactory = $streamFactory ?? $factory; @@ -58,6 +60,12 @@ public function supportsBoundedRequests(): bool return $this->client instanceof GuzzleClient && extension_loaded('curl'); } + /** @internal Custom HTTP clients require an application bootstrap in a fresh process. */ + public function supportsDefaultActivityLeaseProcess(): bool + { + return $this->defaultProcessConfiguration && $this->supportsBoundedRequests(); + } + public function sendBounded(string $method, string $uri, array $headers, ?array $body, int $timeoutSeconds): ?array { if (!$this->supportsBoundedRequests() || $timeoutSeconds < 1 || $timeoutSeconds > 65) { diff --git a/src/Worker.php b/src/Worker.php index 905c6a5..6ca412c 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -15,6 +15,7 @@ use DurableWorkflow\Exception\ServerException; use DurableWorkflow\Exception\TransportException; use DurableWorkflow\Worker\ActivityContext; +use DurableWorkflow\Worker\ActivityLeaseProcess; use DurableWorkflow\Worker\ActivityExecutionFailure; use DurableWorkflow\Worker\CapabilityManifest; use DurableWorkflow\Worker\CancellationHistory; @@ -101,6 +102,10 @@ final class Worker private bool $serviceOperationsSupported = false; private bool $queryTaskAuthoritySupported = false; private ?QueryTaskAuthority $queryTaskAuthority = null; + private bool $activityLeaseRenewalSupported = false; + private ?ActivityLeaseProcess $activityLeaseProcess = null; + /** @var (\Closure(): void)|null */ + private ?\Closure $unbindActivityLease = null; private ?CancellationRequest $claimCancellation = null; private ?string $claimDeliveredCancellationId = null; private bool $preparedLocalActivityGroupsSupported = false; @@ -380,6 +385,12 @@ public function run(int $pollTimeoutSeconds = 5): void } $this->registered = true; $this->lastHeartbeatAt = $this->now(); + if ($this->activityLeaseRenewalSupported && $this->activities !== []) { + if (!ActivityLeaseProcess::available()) { + throw new \RuntimeException('Automatic activity lease maintenance requires PHP CLI proc_open.'); + } + $this->client->activityLeaseProcessConfiguration(); + } $this->diagnostic('worker.registered', [ 'worker_id' => $this->workerId, 'task_queue' => $this->taskQueue, @@ -399,6 +410,7 @@ public function run(int $pollTimeoutSeconds = 5): void ], 'error'); throw $exception; } finally { + $this->stopActivityLease(); $this->closeWorkerSessions(); $this->stickyCache->clear(); if ($this->registered) { @@ -433,6 +445,7 @@ private function captureRegistrationFence(array $registration): void throw new ServerException('The worker registration fencing contract is invalid.', 201, 'invalid_worker_registration_fence'); } + $this->activityLeaseRenewalSupported = ($capabilities['activity_lease_renewal'] ?? null) === true; $capability = $capabilities['worker_deregistration_fencing'] ?? null; if (($capability === null || (is_array($capability) && ($capability['supported'] ?? null) === false)) && $token === null) { @@ -884,6 +897,9 @@ private function pollWithRetry(string $taskKind, \Closure $poll, int $requestedT return $poll($requestId, $timeoutSeconds); } catch (WorkflowClaimRevoked $exception) { + if ($taskKind === 'activity') { + $this->stopActivityLease(); + } if ($taskKind === 'query') { $this->queryTaskAuthority = null; } @@ -896,6 +912,9 @@ private function pollWithRetry(string $taskKind, \Closure $poll, int $requestedT return ['task' => null, 'poll_status' => 'empty']; } catch (ServerException $exception) { + if ($taskKind === 'activity') { + $this->stopActivityLease(); + } $databaseUnavailable = $this->isTransientDatabaseFailure( $exception, "poll_{$taskKind}_task", @@ -940,6 +959,11 @@ private function pollWithRetry(string $taskKind, \Closure $poll, int $requestedT // other task kinds a turn instead of retrying only this one. return $this->shutdownRequested ? null : $exception->details; } + } catch (Throwable $exception) { + if ($taskKind === 'activity') { + $this->stopActivityLease(); + } + throw $exception; } } @@ -1162,13 +1186,19 @@ private function taskFromPoll(array $response): ?array /** @param array $task */ private function executePolledTask(string $taskKind, array $task): void { - if ($this->telemetry === null || $this->client->observabilityContextAvailable()) { - $this->executeTask($taskKind, $task); + try { + if ($this->telemetry === null || $this->client->observabilityContextAvailable()) { + $this->executeTask($taskKind, $task); - return; + return; + } + $this->telemetry->task($taskKind, $task, $this->client->namespace, $this->taskQueue, + fn () => $this->executeTask($taskKind, $task)); + } finally { + if ($taskKind === 'activity') { + $this->stopActivityLease(); + } } - $this->telemetry->task($taskKind, $task, $this->client->namespace, $this->taskQueue, - fn () => $this->executeTask($taskKind, $task)); } /** @param array $task */ @@ -1488,6 +1518,19 @@ private function workerPayloadProgress(string $path, ?array $body, array $respon // progress and cannot renew an activity or query execution budget. $task = $response['task']; $activity = $path === '/worker/activity-tasks/poll'; + if ($activity && $this->activityLeaseRenewalSupported) { + if ($this->activityLeaseProcess !== null) { + throw new WorkflowClaimRevoked('activity_supervisor_conflict', 'A previous activity supervisor is still active.'); + } + $this->activityLeaseProcess = new ActivityLeaseProcess($this->client, $task, $this->workerId, + $this->taskQueue, $response['server_time'] ?? null, hrtime(true) / 1e9, + $this->registered, $this->heartbeatIntervalSeconds); + $this->unbindActivityLease = $this->client->bindActivityLease($this->activityLeaseProcess); + + $process = $this->activityLeaseProcess; + + return static fn () => $process->check(); + } $queryAuthority = null; if (!$activity) { $this->queryTaskAuthority = $queryAuthority = $this->queryAuthorityForTask($task); @@ -2396,6 +2439,7 @@ private function executeActivityTask(array $task): void ), ); $arguments = $this->decodeArguments($task['arguments'] ?? null); + $this->activityLeaseProcess?->check(); if ($this->enableCooperativeCancellation) { $nextObservation = 0.0; $check = function (bool $force) use ($task, &$nextObservation): void { @@ -2458,12 +2502,21 @@ static function () use (&$callbackStopped): void { $callbackStopped = true; }, if ($exception instanceof ServerException && $this->discardClosedActivityAttempt($task, $exception)) { return; } - if ($this->enableCooperativeCancellation && $exception instanceof WorkflowClaimAborted) { + if ($this->activityLeaseProcess !== null && !$exception instanceof WorkflowClaimRevoked) { + try { + $this->activityLeaseProcess->check(); + } catch (WorkflowClaimRevoked $revoked) { + $exception = $revoked; + } + } + if ($exception instanceof WorkflowClaimRevoked + || ($this->enableCooperativeCancellation && $exception instanceof WorkflowClaimAborted)) { if ($callbackStopped) { $this->acknowledgeStoppedRemoteActivity($taskId, $attemptId, $leaseOwner); } $this->diagnostic('worker.claim_aborted', [ 'task_id' => $taskId, 'activity_attempt_id' => $attemptId, + 'reason' => $exception instanceof WorkflowClaimRevoked ? $exception->reason : null, 'task_kind' => 'activity', 'message' => $exception->getMessage(), ], 'warning'); @@ -3502,7 +3555,7 @@ private function preparePoll(int $requestedTimeoutSeconds): int private function heartbeatIfDue(?RequestBudget $budget = null): void { - if (!$this->registered || $this->shutdownRequested) { + if (!$this->registered || $this->shutdownRequested || $this->activityLeaseProcess !== null) { return; } @@ -3513,6 +3566,16 @@ private function heartbeatIfDue(?RequestBudget $budget = null): void $this->heartbeat($budget); } + private function stopActivityLease(): void + { + if ($this->unbindActivityLease !== null) { + ($this->unbindActivityLease)(); + $this->unbindActivityLease = null; + } + $this->activityLeaseProcess?->stop(); + $this->activityLeaseProcess = null; + } + private function heartbeat(?RequestBudget $budget = null): void { if ($this->shutdownRequested || $this->now() < $this->heartbeatRetryAt) { diff --git a/src/Worker/ActivityLeaseAuthority.php b/src/Worker/ActivityLeaseAuthority.php new file mode 100644 index 0000000..9f952b7 --- /dev/null +++ b/src/Worker/ActivityLeaseAuthority.php @@ -0,0 +1,319 @@ + */ + private array $identity; + private float $leaseUntil; + private float $leaseExpiry; + private float $serverTime; + private ?float $businessUntil; + private ?float $heartbeatUntil; + private ?float $heartbeatTime; + private ?float $heartbeatDeadline; + private ?int $heartbeatInterval; + /** @var array{float, float, float}|null */ + private ?array $lastRenewal = null; + private ?ActivityLeaseSession $session; + private ?string $lostReason = null; + private readonly Closure $clock; + + /** + * @param array $task + * @param (Closure(): float)|null $clock Monotonic seconds, including in tests. + */ + public function __construct( + array $task, + string $namespace, + string $owner, + string $queue, + mixed $serverTime, + private readonly float $capturedAt, + ?Closure $clock = null, + ) { + $this->clock = $clock ?? static fn (): float => hrtime(true) / 1e9; + try { + foreach (['task_id', 'workflow_id', 'run_id', 'activity_execution_id', 'activity_attempt_id', + 'lease_owner', 'task_queue'] as $key) { + if (!is_string($task[$key] ?? null) || trim($task[$key]) === '' || strlen($task[$key]) > 255) { + throw new \UnexpectedValueException('The delivery lacks its original '.$key.'.'); + } + } + if ($task['lease_owner'] !== $owner || $task['task_queue'] !== $queue + || ($task['namespace'] ?? $namespace) !== $namespace + || !is_int($task['attempt_number'] ?? null) || $task['attempt_number'] < 1 + || !is_finite($capturedAt) || $capturedAt < 0 || $capturedAt > ($this->clock)()) { + throw new \UnexpectedValueException('The delivery does not belong to this Worker.'); + } + $this->identity = ['task_id' => $task['task_id'], 'workflow_task_id' => $task['task_id'], + 'workflow_instance_id' => $task['workflow_id'], 'workflow_run_id' => $task['run_id'], + 'activity_execution_id' => $task['activity_execution_id'], + 'activity_attempt_id' => $task['activity_attempt_id'], 'attempt_number' => $task['attempt_number'], + 'lease_owner' => $owner]; + $this->serverTime = self::timestamp($serverTime); + $this->leaseExpiry = self::timestamp($task['lease_expires_at'] ?? null); + $this->leaseUntil = $capturedAt + self::duration($this->leaseExpiry, $this->serverTime); + $deadlines = $task['deadlines'] ?? []; + if (!is_array($deadlines)) { + throw new \UnexpectedValueException('Invalid activity deadlines.'); + } + $bounds = []; + foreach (['start_to_close', 'schedule_to_close', 'run', 'execution'] as $key) { + if (isset($deadlines[$key])) { + $bounds[] = $capturedAt + self::timestamp($deadlines[$key]) - $this->serverTime; + } + } + $this->businessUntil = $bounds === [] ? null : min($bounds); + $policy = $task['retry_policy'] ?? null; + $interval = is_array($policy) ? ($policy['heartbeat_timeout'] ?? null) : null; + $this->heartbeatInterval = is_int($interval) && $interval > 0 ? $interval : null; + $this->heartbeatDeadline = isset($deadlines['heartbeat']) ? self::timestamp($deadlines['heartbeat']) : null; + if ($this->heartbeatDeadline !== null && ($this->heartbeatInterval === null + || $this->heartbeatDeadline - $this->serverTime > $this->heartbeatInterval)) { + throw new \UnexpectedValueException('Missing original heartbeat policy.'); + } + $this->heartbeatUntil = $this->heartbeatDeadline === null ? null + : $capturedAt + $this->heartbeatDeadline - $this->serverTime; + $this->heartbeatTime = $this->heartbeatDeadline === null ? null + : $this->heartbeatDeadline - $this->heartbeatInterval; + $this->session = isset($task['worker_session']) + ? new ActivityLeaseSession($task['worker_session'], $namespace, $owner, $this->serverTime, $capturedAt) + : null; + } catch (Throwable $error) { + throw new WorkflowClaimRevoked('invalid_activity_claim', 'Activity delivery lacks valid original authority.', previous: $error); + } + $this->remaining(); + } + + public function remaining(): float + { + return $this->remainingAt(($this->clock)()); + } + + public function fixedBound(): float + { + return min($this->businessUntil ?? INF, $this->session?->ttlBound() ?? INF); + } + + /** @return array */ + public function identity(): array + { + return $this->identity; + } + + /** @return array|null */ + public function session(): ?array + { + return $this->session?->snapshot(); + } + + /** + * A successful HTTP response is insufficient: it must acknowledge this + * original attempt and must arrive before the preceding authority expires. + * @param array $reply + */ + public function acknowledge(array $reply, float $started): void + { + $received = ($this->clock)(); + $this->remainingAt($received); + try { + $this->assertRoundTrip($started, $received); + foreach ($this->identity as $key => $value) { + if (($reply[$key] ?? null) !== $value) { + throw new \UnexpectedValueException('Activity renewal changed '.$key.'.'); + } + } + if (($reply['lease_renewed'] ?? null) !== true || ($reply['heartbeat_recorded'] ?? null) !== false) { + throw new \UnexpectedValueException('Runtime renewal did not acknowledge a metadata-only grant.'); + } + $this->assertContinuation($reply); + $server = self::timestamp($reply['server_time'] ?? null); + $expiry = self::timestamp($reply['lease_expires_at'] ?? null); + $until = $started + self::duration($expiry, $server); + if ($until <= $received) { + throw new \UnexpectedValueException('The renewed lease was consumed by transport.'); + } + // An application heartbeat can overtake an in-flight runtime ping. + // That older ping grants nothing and must not undo actual progress. + if ($this->heartbeatTime !== null && $server < $this->heartbeatTime && $expiry <= $this->leaseExpiry) { + return; + } + if ($server < $this->serverTime || $expiry < $this->leaseExpiry) { + throw new \UnexpectedValueException('Renewal predates the confirmed activity grant.'); + } + $confirmation = [$server, $expiry, $started]; + if ($this->lastRenewal !== null && $server <= $this->lastRenewal[0]) { + if ($confirmation === $this->lastRenewal) { + return; + } + throw new \UnexpectedValueException('Replayed activity renewal.'); + } + $this->observeProgress($reply, $server, $started); + if ($this->session !== null) { + if (!is_array($reply['worker_session'] ?? null)) { + throw new \UnexpectedValueException('Required session grant is missing from renewal.'); + } + $this->session->acknowledge($reply['worker_session'], $server, $started); + } + $this->leaseUntil = max($this->leaseUntil, $until); + $this->serverTime = $server; + $this->leaseExpiry = $expiry; + $this->lastRenewal = $confirmation; + } catch (Throwable $error) { + $this->lose('invalid_activity_renewal', $error); + } + $this->remaining(); + } + + /** + * Only an actual user heartbeat can move the recorded progress deadline. + * The receipt instant is retained when the parent relays it over private IPC. + * @param array $reply + */ + public function acknowledgeProgress(array $reply, float $started, ?float $received = null): void + { + $received ??= ($this->clock)(); + $this->remainingAt($received); + try { + $this->assertRoundTrip($started, $received); + foreach (['task_id', 'activity_attempt_id', 'lease_owner'] as $key) { + if (($reply[$key] ?? null) !== $this->identity[$key]) { + throw new \UnexpectedValueException('Heartbeat identifies another activity.'); + } + } + $this->assertContinuation($reply); + if (($reply['heartbeat_recorded'] ?? null) !== true) { + throw new \UnexpectedValueException('No application heartbeat was recorded.'); + } + $progress = self::timestamp($reply['last_heartbeat_at'] ?? null); + $expiry = self::timestamp($reply['lease_expires_at'] ?? null); + $until = $started + self::duration($expiry, $progress); + if ($until <= $received) { + throw new \UnexpectedValueException('The heartbeat grant was consumed by transport.'); + } + if ($this->heartbeatTime !== null && $progress <= $this->heartbeatTime) { + return; + } + if ($this->heartbeatInterval !== null) { + $deadline = self::timestamp($reply['heartbeat_deadline_at'] ?? null); + if (abs($deadline - $progress - $this->heartbeatInterval) > 0.000001) { + throw new \UnexpectedValueException('Heartbeat changed the recorded progress interval.'); + } + $this->heartbeatUntil = max($this->heartbeatUntil ?? 0, $started + $this->heartbeatInterval); + $this->heartbeatDeadline = $deadline; + } + $this->heartbeatTime = $progress; + $this->leaseUntil = max($this->leaseUntil, $until); + $this->leaseExpiry = max($this->leaseExpiry, $expiry); + } catch (Throwable $error) { + $this->lose('invalid_activity_heartbeat', $error); + } + $this->remaining(); + } + + /** @param array $reply */ + public function acknowledgeSession(array $reply, float $started): void + { + $this->remaining(); + try { + $this->assertRoundTrip($started, ($this->clock)()); + if ($this->session === null) { + throw new \UnexpectedValueException('No original session was required.'); + } + $this->session->acknowledge($reply, self::timestamp($reply['server_time'] ?? null), $started); + } catch (Throwable $error) { + $this->lose('invalid_activity_session', $error); + } + $this->remaining(); + } + + private function remainingAt(float $now): float + { + if ($this->lostReason === null) { + $remaining = min($this->leaseUntil, $this->businessUntil ?? INF, + $this->heartbeatUntil ?? INF, $this->session?->bound() ?? INF) - $now; + if ($remaining > 0) { + return $remaining; + } + $this->lostReason = 'activity_execution_budget_expired'; + } + throw new WorkflowClaimRevoked($this->lostReason, 'The original activity execution authority is unavailable.', + taskId: (string) $this->identity['task_id']); + } + + /** @param array $reply */ + private function assertContinuation(array $reply): void + { + if (($reply['can_continue'] ?? null) !== true || ($reply['cancel_requested'] ?? null) !== false + || ($reply['reason'] ?? null) !== null) { + throw new \UnexpectedValueException('The Server refused the original activity claim.'); + } + } + + private function assertRoundTrip(float $started, float $received): void + { + if (!is_finite($started) || !is_finite($received) || $started < $this->capturedAt + || $started > $received || $received > ($this->clock)()) { + throw new \UnexpectedValueException('Invalid monotonic activity request timing.'); + } + } + + /** @param array $reply */ + private function observeProgress(array $reply, float $server, float $started): void + { + if ($this->heartbeatInterval === null || $this->heartbeatUntil === null) { + return; + } + $progress = self::timestamp($reply['last_heartbeat_at'] ?? null); + $deadline = self::timestamp($reply['heartbeat_deadline_at'] ?? null); + if ($progress > $server || abs($deadline - $progress - $this->heartbeatInterval) > 0.000001) { + throw new \UnexpectedValueException('Runtime renewal changed application progress policy.'); + } + $bound = $started + $deadline - $server; + $this->heartbeatUntil = $progress > $this->heartbeatTime + ? max($this->heartbeatUntil, $bound) : min($this->heartbeatUntil, $bound); + $this->heartbeatTime = max($this->heartbeatTime ?? 0, $progress); + $this->heartbeatDeadline = $deadline; + } + + private function lose(string $reason, Throwable $error): never + { + $this->lostReason = $reason; + throw new WorkflowClaimRevoked($reason, 'Activity continuation could not be confirmed.', previous: $error, + taskId: (string) $this->identity['task_id']); + } + + /** @internal Canonical UTC6 timestamps cannot normalize an invalid calendar date. */ + public static function timestamp(mixed $value): float + { + if (!is_string($value) || preg_match('/\A\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z\z/', $value) !== 1) { + throw new \UnexpectedValueException('Activity timing must be UTC with six fractional digits.'); + } + $date = DateTimeImmutable::createFromFormat('!Y-m-d\TH:i:s.u\Z', $value, new DateTimeZone('UTC')); + if ($date === false || $date->format('Y-m-d\TH:i:s.u\Z') !== $value) { + throw new \UnexpectedValueException('Activity timing contains an invalid calendar date.'); + } + + return (float) $date->format('U.u'); + } + + private static function duration(float $expiry, float $server): float + { + $duration = $expiry - $server; + if ($duration <= 0 || $duration > 86400) { + throw new \UnexpectedValueException('Activity lease duration is invalid.'); + } + + return $duration; + } +} diff --git a/src/Worker/ActivityLeaseBinding.php b/src/Worker/ActivityLeaseBinding.php new file mode 100644 index 0000000..2a38c68 --- /dev/null +++ b/src/Worker/ActivityLeaseBinding.php @@ -0,0 +1,50 @@ + Only Fiber objects are used as keys. */ + private WeakMap $fibers; + private ?ActivityLeaseProcess $main = null; + + public function __construct() + { + $this->fibers = new WeakMap(); + } + + public function current(): ?ActivityLeaseProcess + { + $fiber = Fiber::getCurrent(); + + return $fiber === null ? $this->main : ($this->fibers[$fiber] ?? null); + } + + public function enter(ActivityLeaseProcess $process): Closure + { + if ($this->current() !== null) { + throw new \LogicException('An activity execution context already owns a lease supervisor.'); + } + $fiber = Fiber::getCurrent(); + if ($fiber === null) { + $this->main = $process; + } else { + $this->fibers[$fiber] = $process; + } + + return function () use ($fiber, $process): void { + if ($fiber === null && $this->main === $process) { + $this->main = null; + } elseif ($fiber !== null && ($this->fibers[$fiber] ?? null) === $process) { + unset($this->fibers[$fiber]); + } + }; + } +} diff --git a/src/Worker/ActivityLeaseMaintainer.php b/src/Worker/ActivityLeaseMaintainer.php new file mode 100644 index 0000000..5891c75 --- /dev/null +++ b/src/Worker/ActivityLeaseMaintainer.php @@ -0,0 +1,249 @@ + $start */ + private function __construct(private readonly array $start) + { + $configuration = $start['client']; + if (!is_array($configuration) || !is_array($start['task'] ?? null) + || !is_string($start['owner'] ?? null) || !is_string($start['queue'] ?? null) + || !is_numeric($start['captured_at'] ?? null) || !is_int($start['parent_pid'] ?? null)) { + throw new \UnexpectedValueException('Malformed private activity maintenance configuration.'); + } + $this->client = self::client($configuration)->withBoundedWorkerRequests(); + $this->authority = new ActivityLeaseAuthority($start['task'], $this->client->namespace, + $start['owner'], $start['queue'], $start['server_time'] ?? null, (float) $start['captured_at']); + } + + public static function run(): void + { + $first = fgets(STDIN, 65537); + $start = []; + try { + if ($first === false || !str_ends_with($first, "\n")) { + return; + } + $start = json_decode($first, true, 32, JSON_THROW_ON_ERROR); + if (!is_array($start) || ($start['operation'] ?? null) !== 'start') { + return; + } + $maintainer = new self($start); + stream_set_blocking(STDIN, false); + $maintainer->maintainInitialGrant(); + $maintainer->respond($start); + $maintainer->loop(); + } catch (Throwable) { + // Bootstrap exceptions can contain application secrets. Only a + // bounded protocol reason is ever written to the parent pipe. + self::write(['id' => is_array($start) ? ($start['id'] ?? 0) : 0, + 'can_continue' => false, 'reason' => 'activity_maintenance_bootstrap_failed']); + } + } + + /** @param array $configuration */ + private static function client(array $configuration): Client + { + if (!is_string($configuration['base_uri'] ?? null) || !is_string($configuration['namespace'] ?? null) + || !is_string($configuration['worker_protocol'] ?? null)) { + throw new \UnexpectedValueException('Missing maintenance runtime identity.'); + } + $bootstrap = $configuration['bootstrap'] ?? null; + if ($bootstrap !== null) { + if (!is_string($bootstrap)) { + throw new \UnexpectedValueException('Malformed activity bootstrap.'); + } + ob_start(); + try { + $client = require $bootstrap; + } finally { + $output = ob_get_clean(); + } + if ($output !== '' || !$client instanceof Client + || $client->activityLeaseEndpoint() !== $configuration['base_uri'] + || $client->namespace !== $configuration['namespace'] + || $client->workerProtocolVersion !== $configuration['worker_protocol']) { + throw new \UnexpectedValueException('Bootstrap changed the maintenance runtime identity.'); + } + + return $client; + } + $headers = $configuration['worker_headers'] ?? null; + if (!is_array($headers)) { + throw new \UnexpectedValueException('Missing original worker authentication.'); + } + foreach ($headers as $key => $value) { + if (!is_string($key) || !is_string($value)) { + throw new \UnexpectedValueException('Malformed original worker authentication.'); + } + } + + return new Client($configuration['base_uri'], new StaticHeadersAuthentication($headers), + namespace: $configuration['namespace'], workerProtocolVersion: $configuration['worker_protocol']); + } + + private function loop(): void + { + while (!feof(STDIN) && $this->parentPresent()) { + $now = hrtime(true) / 1e9; + $reply = null; + $started = $now; + if ($this->lostReason === null && $now >= $this->nextRenewal) { + try { + $reply = $this->requestRenewal(); + } catch (Throwable $error) { + $this->recordMaintenanceError($error); + } + } + // Actual parent progress can overtake the in-flight runtime reply. + // Drain those receipts before evaluating the old heartbeat ceiling. + $commands = $this->readCommands(); + foreach ($commands as $command) { + if (($command['operation'] ?? null) === 'progress' && $this->lostReason === null) { + try { + $this->authority->acknowledgeProgress($command['reply'], + $command['started'], $command['received']); + } catch (Throwable $error) { + $this->recordMaintenanceError($error); + } + } + } + if ($reply !== null && $this->lostReason === null) { + try { + $this->authority->acknowledge($reply, $started); + $this->scheduleRenewal(); + } catch (Throwable $error) { + $this->recordMaintenanceError($error); + } + } + foreach ($commands as $command) { + if (!in_array($command['operation'] ?? null, ['check', 'progress'], true)) { + $this->lostReason = 'activity_maintenance_invalid_command'; + } + $this->respond($command); + } + usleep(10000); + } + } + + private function maintainInitialGrant(): void + { + $started = hrtime(true) / 1e9; + $this->authority->acknowledge($this->requestRenewal(), $started); + $this->scheduleRenewal(); + } + + /** @return array */ + private function requestRenewal(): array + { + if (!$this->parentPresent()) { + throw new WorkflowClaimRevoked('activity_parent_stopped', 'Original activity process has stopped.'); + } + $identity = $this->authority->identity(); + if (($this->start['registered'] ?? false) === true && hrtime(true) / 1e9 >= $this->nextPresence) { + $ack = $this->client->heartbeatWorker((string) $identity['lease_owner'], + ['workflow_available' => 0, 'activity_available' => 0, 'session_available' => 0], $this->budget()); + if (($ack['worker_id'] ?? null) !== $identity['lease_owner'] || ($ack['acknowledged'] ?? null) !== true) { + throw new WorkflowClaimRevoked('activity_worker_unconfirmed', 'Original Worker presence was not acknowledged.'); + } + $interval = $this->start['heartbeat_interval'] ?? 10; + $this->nextPresence = hrtime(true) / 1e9 + max(1, min(10, is_int($interval) ? $interval : 10)); + } + + return $this->client->renewActivityTaskLease((string) $identity['task_id'], + (string) $identity['activity_attempt_id'], (string) $identity['lease_owner'], $this->budget()); + } + + private function budget(): RequestBudget + { + return new RequestBudget(5, hrtime(true) / 1e9 + $this->authority->remaining()); + } + + private function scheduleRenewal(): void + { + $this->nextRenewal = hrtime(true) / 1e9 + min(10, $this->authority->remaining() / 3); + } + + private function recordMaintenanceError(Throwable $error): void + { + if ($error instanceof ServerException && ($error->isTransientConnectionFailure() + || $error->isTransientUpstreamFailure() || $error->isStorageAdmissionFailure())) { + // Unavailability adds no time. Retry only within the preceding grant. + $this->nextRenewal = hrtime(true) / 1e9 + 0.1; + + return; + } + $this->lostReason = $error instanceof WorkflowClaimRevoked ? $error->reason : 'activity_maintenance_refused'; + } + + /** @return list> */ + private function readCommands(): array + { + $part = stream_get_contents(STDIN, 65537 - strlen($this->buffer)); + if ($part === false || strlen($part) + strlen($this->buffer) > 65536) { + throw new \UnexpectedValueException('Activity maintenance input exceeded its private bound.'); + } + $this->buffer .= $part; + $commands = []; + while (str_contains($this->buffer, "\n")) { + [$line, $this->buffer] = explode("\n", $this->buffer, 2); + $command = json_decode($line, true, 32, JSON_THROW_ON_ERROR); + if (!is_array($command)) { + throw new \UnexpectedValueException('Invalid private activity command.'); + } + $commands[] = $command; + } + + return $commands; + } + + /** @param array $command */ + private function respond(array $command): void + { + try { + if ($this->lostReason !== null) { + throw new WorkflowClaimRevoked($this->lostReason, 'Original activity grant is unavailable.'); + } + $now = hrtime(true) / 1e9; + $remaining = $this->authority->remaining(); + self::write(['id' => $command['id'] ?? 0, 'can_continue' => true, + 'until' => $now + $remaining, 'monotonic' => $now]); + } catch (Throwable $error) { + $this->recordMaintenanceError($error); + self::write(['id' => $command['id'] ?? 0, 'can_continue' => false, 'reason' => $this->lostReason]); + } + } + + private function parentPresent(): bool + { + return !function_exists('posix_getppid') || posix_getppid() === $this->start['parent_pid']; + } + + /** @param array $message */ + private static function write(array $message): void + { + $line = json_encode($message, JSON_THROW_ON_ERROR)."\n"; + if (fwrite(STDOUT, $line) !== strlen($line)) { + throw new \RuntimeException('Original activity process is unavailable.'); + } + fflush(STDOUT); + } +} diff --git a/src/Worker/ActivityLeaseProcess.php b/src/Worker/ActivityLeaseProcess.php new file mode 100644 index 0000000..57f1282 --- /dev/null +++ b/src/Worker/ActivityLeaseProcess.php @@ -0,0 +1,228 @@ + */ + private array $pipes = []; + private string $buffer = ''; + private int $sequence = 0; + private int $parentPid = 0; + private ?string $lostReason = null; + private float $until; + private readonly float $fixedUntil; + /** @var array */ + private readonly array $identity; + + /** @param array $task */ + public function __construct(Client $client, array $task, string $owner, string $queue, + mixed $serverTime, float $captured, bool $registered, int $heartbeatInterval) + { + $this->parentPid = getmypid(); + $authority = new ActivityLeaseAuthority($task, $client->namespace, $owner, $queue, $serverTime, $captured); + $this->identity = $authority->identity(); + $this->until = hrtime(true) / 1e9 + $authority->remaining(); + $this->fixedUntil = $authority->fixedBound(); + if (!self::available()) { + throw new \RuntimeException('Automatic activity lease maintenance requires PHP CLI proc_open.'); + } + $configuration = $client->activityLeaseProcessConfiguration(); + $autoload = null; + foreach (ClassLoader::getRegisteredLoaders() as $directory => $loader) { + if (realpath((string) $loader->findFile(Client::class)) === realpath(__DIR__.'/../Client.php')) { + $autoload = $directory.'/autoload.php'; + break; + } + } + if ($autoload === null) { + throw new \RuntimeException('Activity maintenance requires this SDK\'s Composer autoloader.'); + } + $descriptors = [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], + 2 => ['file', PHP_OS_FAMILY === 'Windows' ? 'NUL' : '/dev/null', 'w']]; + $this->process = proc_open([PHP_BINARY, __DIR__.'/activity-lease-maintainer.php', $autoload], + $descriptors, $this->pipes, options: ['bypass_shell' => true]); + if (!is_resource($this->process)) { + throw new \RuntimeException('Could not start independent activity lease maintenance.'); + } + foreach ($this->pipes as $pipe) { + stream_set_blocking($pipe, false); + } + try { + $this->exchange('start', ['client' => $configuration, 'task' => array_intersect_key($task, + array_flip(['task_id', 'workflow_id', 'run_id', 'activity_execution_id', 'activity_attempt_id', + 'attempt_number', 'lease_owner', 'task_queue', 'namespace', 'lease_expires_at', + 'deadlines', 'retry_policy', 'worker_session'])), + 'server_time' => $serverTime, 'captured_at' => $captured, 'owner' => $owner, 'queue' => $queue, + 'parent_pid' => getmypid(), 'registered' => $registered, 'heartbeat_interval' => $heartbeatInterval]); + } catch (Throwable $error) { + $this->stop(); + throw $error; + } + } + + public function governs(string $path): bool + { + return in_array($path, array_map(fn (string $action): string => '/worker/activity-tasks/' + .rawurlencode((string) $this->identity['task_id']).'/'.$action, + ['complete', 'fail', 'heartbeat', 'renew-lease']), true); + } + + /** @param array|null $body */ + public function checkRequest(string $path, ?array $body): void + { + if (($body['activity_attempt_id'] ?? null) !== $this->identity['activity_attempt_id'] + || ($body['lease_owner'] ?? null) !== $this->identity['lease_owner']) { + throw new WorkflowClaimRevoked('activity_identity_mismatch', 'Activity publication changed its original claim.'); + } + $this->check(); + } + + public function check(): void + { + $this->exchange('check'); + } + + public function budget(): RequestBudget + { + return new RequestBudget(30, min($this->until, $this->fixedUntil)); + } + + /** @param array $reply */ + public function progress(array $reply, float $started, float $received): void + { + $this->exchange('progress', ['reply' => $reply, 'started' => $started, 'received' => $received]); + } + + public function stop(): void + { + foreach ($this->pipes as $pipe) { + if (is_resource($pipe)) { + fclose($pipe); + } + } + $this->pipes = []; + if ($this->parentPid !== 0 && $this->parentPid !== getmypid()) { + // An explicitly isolated callback may inherit descriptors, but may + // never terminate or converse with its parent's maintenance helper. + $this->process = null; + + return; + } + if (is_resource($this->process)) { + $until = hrtime(true) / 1e9 + 0.25; + while ($this->running() && hrtime(true) / 1e9 < $until) { + usleep(5000); + } + if ($this->running()) { + // This stops maintenance only. It is never callback-stop proof. + proc_terminate($this->process); + $until = hrtime(true) / 1e9 + 0.25; + while ($this->running() && hrtime(true) / 1e9 < $until) { + usleep(5000); + } + if ($this->running()) { + proc_terminate($this->process, 9); + } + } + proc_close($this->process); + } + $this->process = null; + } + + /** @phpstan-impure Process status changes independently of this PHP process. */ + private function running(): bool + { + return is_resource($this->process) && proc_get_status($this->process)['running']; + } + + public function __destruct() + { + $this->stop(); + } + + /** @param array $body */ + private function exchange(string $operation, array $body = []): void + { + if ($this->parentPid !== getmypid()) { + throw new WorkflowClaimRevoked('activity_supervisor_process_mismatch', + 'Only the original Worker process can use its activity maintenance channel.'); + } + if ($this->lostReason !== null) { + $this->refuse($this->lostReason); + } + $started = hrtime(true) / 1e9; + if ($started >= $this->fixedUntil || ($operation === 'start' && $started >= $this->until)) { + $this->refuse('activity_execution_budget_expired'); + } + // The helper can renew a runtime grant while the inline callback is + // blocked. Obtain current proof; immutable business/TTL bounds remain. + $deadline = min($this->fixedUntil, $started + 5, $operation === 'start' ? $this->until : INF); + $id = ++$this->sequence; + $line = json_encode(['id' => $id, 'operation' => $operation] + $body, JSON_THROW_ON_ERROR)."\n"; + if (strlen($line) > 65536) { + $this->refuse('activity_maintenance_message_too_large'); + } + try { + $offset = 0; + while ($offset < strlen($line)) { + $written = fwrite($this->pipes[0], substr($line, $offset)); + if ($written === false || hrtime(true) / 1e9 >= $deadline) { + $this->refuse('activity_maintenance_unavailable'); + } + $offset += $written; + if ($written === 0) { + usleep(1000); + } + } + while (!str_contains($this->buffer, "\n")) { + $part = stream_get_contents($this->pipes[1], 65537 - strlen($this->buffer)); + if ($part === false || feof($this->pipes[1]) || strlen($this->buffer) + strlen($part) > 65536 + || hrtime(true) / 1e9 >= $deadline) { + $this->refuse('activity_maintenance_unavailable'); + } + $this->buffer .= $part; + usleep(1000); + } + [$raw, $this->buffer] = explode("\n", $this->buffer, 2); + $reply = json_decode($raw, true, 32, JSON_THROW_ON_ERROR); + $now = hrtime(true) / 1e9; + if (!is_array($reply) || ($reply['id'] ?? null) !== $id || ($reply['can_continue'] ?? null) !== true + || !is_numeric($reply['until'] ?? null) || !is_numeric($reply['monotonic'] ?? null) + || $reply['monotonic'] < $started || $reply['monotonic'] > $now + || $reply['until'] <= $now || $now >= $deadline) { + $reason = is_array($reply) && is_string($reply['reason'] ?? null) + ? $reply['reason'] : 'activity_maintenance_unconfirmed'; + $this->refuse($reason); + } + $this->until = (float) $reply['until']; + } catch (WorkflowClaimRevoked $error) { + throw $error; + } catch (Throwable $error) { + $this->lostReason = 'activity_maintenance_unavailable'; + throw new WorkflowClaimRevoked($this->lostReason, 'Independent activity maintenance became unavailable.', previous: $error); + } + } + + private function refuse(string $reason): never + { + $this->lostReason = $reason; + throw new WorkflowClaimRevoked($reason, 'The original activity lease is unavailable.', + taskId: (string) $this->identity['task_id']); + } +} diff --git a/src/Worker/ActivityLeaseSession.php b/src/Worker/ActivityLeaseSession.php new file mode 100644 index 0000000..6e1163a --- /dev/null +++ b/src/Worker/ActivityLeaseSession.php @@ -0,0 +1,108 @@ + */ + private array $value; + /** @var array */ + private array $options; + private float $ttl; + private float $ttlUntil; + private float $leaseExpiry; + private float $leaseUntil; + private float $serverTime; + + public function __construct(mixed $value, private readonly string $namespace, private readonly string $owner, + float $server, float $started) + { + if (!is_array($value)) { + throw new \UnexpectedValueException('Invalid required activity session.'); + } + $this->ttl = ActivityLeaseAuthority::timestamp($value['ttl_expires_at'] ?? null); + $this->leaseExpiry = ActivityLeaseAuthority::timestamp($value['lease_expires_at'] ?? null); + $this->ttlUntil = $started + $this->ttl - $server; + $this->leaseUntil = $started + $this->leaseExpiry - $server; + $this->serverTime = $server; + $this->options = self::options($value); + $this->acknowledge($value, $server, $started); + } + + public function bound(): float + { + return min($this->ttlUntil, $this->leaseUntil); + } + + public function ttlBound(): float + { + return $this->ttlUntil; + } + + /** @return array */ + public function snapshot(): array + { + return $this->value; + } + + /** @param array $value */ + public function acknowledge(array $value, float $server, float $started): void + { + $expiry = ActivityLeaseAuthority::timestamp($value['lease_expires_at'] ?? null); + if (self::options($value) !== $this->options || ($value['status'] ?? null) !== 'active' + || ($value['lease_owner'] ?? null) !== $this->owner + || ($value['namespace'] ?? $this->namespace) !== $this->namespace + || ActivityLeaseAuthority::timestamp($value['ttl_expires_at'] ?? null) !== $this->ttl + || $expiry < $this->leaseExpiry || $server < $this->serverTime) { + throw new \UnexpectedValueException('Session renewal changed the original holder, policy or TTL.'); + } + $this->ttlUntil = min($this->ttlUntil, $started + $this->ttl - $server); + // Re-reading the same grant is an observation, not a lease extension. + $this->leaseUntil = $expiry > $this->leaseExpiry + ? max($this->leaseUntil, $started + $expiry - $server) + : min($this->leaseUntil, $started + $expiry - $server); + $this->leaseExpiry = $expiry; + $this->serverTime = $server; + $this->value = $value; + } + + /** @param array $value + * @return array + */ + private static function options(array $value): array + { + $required = ['session_id', 'lease_seconds', 'ttl_seconds', 'max_concurrent_activities', + 'create_if_missing', 'allow_reacquire_after_failure']; + foreach ($required as $key) { + if (!array_key_exists($key, $value)) { + throw new \UnexpectedValueException('Missing original session option '.$key.'.'); + } + } + if (!is_string($value['session_id']) || trim($value['session_id']) === '' + || !is_array($value['requirements'] ?? null) + || (isset($value['queue']) && !is_string($value['queue'])) + || !is_bool($value['create_if_missing']) || !is_bool($value['allow_reacquire_after_failure'])) { + throw new \UnexpectedValueException('Malformed original session options.'); + } + foreach (['lease_seconds', 'ttl_seconds', 'max_concurrent_activities'] as $key) { + if (!is_int($value[$key]) || $value[$key] < 1) { + throw new \UnexpectedValueException('Invalid session duration or concurrency.'); + } + } + foreach ($value['requirements'] as $item) { + if (!is_string($item) || trim($item) === '') { + throw new \UnexpectedValueException('Invalid session requirement.'); + } + } + + $options = []; + foreach ([...$required, 'queue', 'requirements'] as $key) { + $options[$key] = $value[$key] ?? null; + } + + return $options; + } +} diff --git a/src/Worker/activity-lease-maintainer.php b/src/Worker/activity-lease-maintainer.php new file mode 100644 index 0000000..7c82c21 --- /dev/null +++ b/src/Worker/activity-lease-maintainer.php @@ -0,0 +1,10 @@ + $this->now); + } + + public function testRenewalChargesNetworkTimeWithoutChangingOriginalIdentity(): void + { + $authority = $this->authority(); + $this->now = 102; + $authority->acknowledge(self::renewal(2, 12), 101); + self::assertSame(9.0, $authority->remaining()); + self::assertSame('0', $authority->identity()['workflow_instance_id']); + self::assertSame('017', $authority->identity()['workflow_run_id']); + $this->now = 109; + self::assertSame(2.0, $authority->remaining()); + $this->now = 111; + $this->expectException(WorkflowClaimRevoked::class); + $authority->remaining(); + } + + #[DataProvider('identityCorruption')] + public function testRenewalRefusesEveryChangedOriginalIdentity(string $key, mixed $value): void + { + $authority = $this->authority(); + $reply = self::renewal(1, 20); + $reply[$key] = $value; + try { + $authority->acknowledge($reply, 100); + self::fail('A different activity identity acquired authority.'); + } catch (WorkflowClaimRevoked $error) { + self::assertSame('invalid_activity_renewal', $error->reason); + } + // A subsequent valid response must not restore a refused claim. + $this->expectException(WorkflowClaimRevoked::class); + $authority->acknowledge(self::renewal(2, 21), 100); + } + + public static function identityCorruption(): iterable + { + foreach (['task_id', 'workflow_task_id', 'workflow_instance_id', 'workflow_run_id', + 'activity_execution_id', 'activity_attempt_id', 'lease_owner'] as $key) { + yield $key => [$key, 'foreign']; + } + yield 'attempt generation' => ['attempt_number', 2]; + yield 'string generation' => ['attempt_number', '1']; + yield 'refused' => ['can_continue', false]; + yield 'truthy' => ['lease_renewed', 1]; + yield 'cancelled' => ['cancel_requested', true]; + yield 'user progress' => ['heartbeat_recorded', true]; + yield 'reason' => ['reason', 'attempt_closed']; + yield 'missing lease' => ['lease_expires_at', null]; + yield 'bad date' => ['server_time', '2026-02-30T00:00:00.000000Z']; + yield 'offset' => ['server_time', '2026-01-01T00:00:00.000000+00:00']; + } + + #[DataProvider('businessDeadlines')] + public function testRenewalCannotExtendAnyOriginalBusinessDeadline(string $name): void + { + $authority = $this->authority(['deadlines' => [$name => self::stamp(3)]]); + $this->now = 102; + $authority->acknowledge(self::renewal(2, 100), 101); + self::assertSame(1.0, $authority->remaining()); + $this->now = 103; + $this->expectException(WorkflowClaimRevoked::class); + $authority->remaining(); + } + + public static function businessDeadlines(): iterable + { + foreach (['start_to_close', 'schedule_to_close', 'run', 'execution'] as $name) { + yield $name => [$name]; + } + } + + public function testSuccessAfterPriorLeaseExpiredCannotReviveAuthority(): void + { + $authority = $this->authority(); + $this->now = 110; + $this->expectException(WorkflowClaimRevoked::class); + $authority->acknowledge(self::renewal(10, 100), 109); + } + + public function testReplayOfSuccessfulRenewalDoesNotResetItsClock(): void + { + $authority = $this->authority(); + $this->now = 101; + $reply = self::renewal(1, 20); + $authority->acknowledge($reply, 100); + $this->now = 102; + $authority->acknowledge($reply, 100); + self::assertSame(17.0, $authority->remaining()); + $this->expectException(WorkflowClaimRevoked::class); + $authority->acknowledge($reply, 102); + } + + public function testRuntimeRenewalDoesNotPretendToBeApplicationProgress(): void + { + $authority = $this->heartbeatAuthority(); + $this->now = 102; + $authority->acknowledge(self::renewal(2, 20) + ['last_heartbeat_at' => self::stamp(0), + 'heartbeat_deadline_at' => self::stamp(4)], 101); + self::assertSame(1.0, $authority->remaining(), 'The full control round trip consumes progress budget.'); + $this->now = 103; + $this->expectException(WorkflowClaimRevoked::class); + $authority->remaining(); + } + + public function testOnlyRecordedProgressWithTheOriginalIntervalExtendsHeartbeatDeadline(): void + { + $authority = $this->heartbeatAuthority(); + $this->now = 102; + $authority->acknowledgeProgress(self::progress(2, 6, 12), 101); + self::assertSame(3.0, $authority->remaining()); + $this->now = 104; + $authority->acknowledge(self::renewal(4, 20) + ['last_heartbeat_at' => self::stamp(2), + 'heartbeat_deadline_at' => self::stamp(6)], 103); + self::assertSame(1.0, $authority->remaining()); + } + + public function testBufferedHeartbeatCannotRenewAnAlreadyExpiredOriginalReceipt(): void + { + $authority = $this->heartbeatAuthority(); + $this->now = 105; + $this->expectException(WorkflowClaimRevoked::class); + $authority->acknowledgeProgress(self::progress(4, 8, 20), 103, 104); + } + + public function testBufferedTimelyProgressIsChargedBeforeCurrentContinuation(): void + { + $authority = $this->heartbeatAuthority(); + $this->now = 104; + $authority->acknowledgeProgress(self::progress(2, 6, 12), 101, 102); + self::assertSame(1.0, $authority->remaining()); + } + + public function testLargerHeartbeatIntervalCannotBeInventedByAReply(): void + { + $authority = $this->heartbeatAuthority(); + $this->now = 102; + $this->expectException(WorkflowClaimRevoked::class); + $authority->acknowledgeProgress(self::progress(2, 8, 12), 101); + } + + public function testRequiredSessionCanRenewOnlyItsLeaseAndNeverItsOriginalTtl(): void + { + $session = self::session(); + $authority = $this->authority(['worker_session' => $session]); + $this->now = 102; + $authority->acknowledgeSession(array_replace($session, ['lease_expires_at' => self::stamp(9), + 'server_time' => self::stamp(2)]), 101); + self::assertSame(4.0, $authority->remaining()); + $this->now = 106; + $this->expectException(WorkflowClaimRevoked::class); + $authority->remaining(); + } + + #[DataProvider('sessionChanges')] + public function testSessionChangeCannotAcquireOrExtendAnotherGrant(string $key, mixed $value): void + { + $authority = $this->authority(['worker_session' => self::session()]); + $this->now = 101; + $reply = array_replace(self::session(), ['server_time' => self::stamp(1), + 'lease_expires_at' => self::stamp(9), $key => $value]); + $this->expectException(WorkflowClaimRevoked::class); + $authority->acknowledgeSession($reply, 100); + } + + public static function sessionChanges(): iterable + { + yield 'TTL' => ['ttl_expires_at', self::stamp(100)]; + yield 'owner' => ['lease_owner', 'another-owner']; + yield 'namespace' => ['namespace', 'another-tenant']; + yield 'id' => ['session_id', 'another-session']; + yield 'queue' => ['queue', 'another-queue']; + yield 'requirements' => ['requirements', ['gpu']]; + yield 'policy' => ['lease_seconds', 100]; + yield 'status' => ['status', 'closed']; + } + + public function testJsonObjectMemberOrderDoesNotChangeSessionIdentity(): void + { + $authority = $this->authority(['worker_session' => self::session()]); + $this->now = 101; + $reply = array_replace(array_reverse(self::session(), true), ['server_time' => self::stamp(1)]); + $authority->acknowledgeSession($reply, 100); + self::assertSame(2.0, $authority->remaining()); + } + + private function heartbeatAuthority(): ActivityLeaseAuthority + { + return $this->authority(['deadlines' => ['heartbeat' => self::stamp(4)], + 'retry_policy' => ['heartbeat_timeout' => 4]]); + } + + private static function task(): array + { + return ['task_id' => 'task', 'workflow_id' => '0', 'run_id' => '017', + 'activity_execution_id' => 'execution', 'activity_attempt_id' => 'attempt', 'attempt_number' => 1, + 'lease_owner' => 'owner', 'task_queue' => 'queue', 'namespace' => 'tenant', + 'lease_expires_at' => self::stamp(10)]; + } + + private static function renewal(int $server, int $expiry): array + { + return ['task_id' => 'task', 'workflow_task_id' => 'task', 'workflow_instance_id' => '0', + 'workflow_run_id' => '017', 'activity_execution_id' => 'execution', 'activity_attempt_id' => 'attempt', + 'attempt_number' => 1, 'lease_owner' => 'owner', 'lease_renewed' => true, 'can_continue' => true, + 'cancel_requested' => false, 'heartbeat_recorded' => false, 'reason' => null, + 'server_time' => self::stamp($server), 'lease_expires_at' => self::stamp($expiry)]; + } + + private static function progress(int $progress, int $deadline, int $expiry): array + { + return array_replace(self::renewal($progress, $expiry), ['heartbeat_recorded' => true, + 'last_heartbeat_at' => self::stamp($progress), 'heartbeat_deadline_at' => self::stamp($deadline)]); + } + + private static function session(): array + { + return ['session_id' => 'session', 'status' => 'active', 'namespace' => 'tenant', 'lease_owner' => 'owner', + 'queue' => 'queue', 'requirements' => [], 'lease_seconds' => 4, 'ttl_seconds' => 7, + 'max_concurrent_activities' => 1, 'create_if_missing' => true, 'allow_reacquire_after_failure' => true, + 'lease_expires_at' => self::stamp(4), 'ttl_expires_at' => self::stamp(7)]; + } + + private static function stamp(int $seconds): string + { + return gmdate('Y-m-d\TH:i:s', 1767225600 + $seconds).'.000000Z'; + } +} diff --git a/tests/Support/activity-lease-http.php b/tests/Support/activity-lease-http.php new file mode 100644 index 0000000..fb59f63 --- /dev/null +++ b/tests/Support/activity-lease-http.php @@ -0,0 +1,99 @@ + (new DateTimeImmutable('@'.sprintf('%.6F', $value))) + ->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d\TH:i:s.u\Z'); +$state['requests'][] = ['path' => $path, 'body' => $body, + 'namespace' => $_SERVER['HTTP_X_NAMESPACE'] ?? null, + 'authorization' => $_SERVER['HTTP_AUTHORIZATION'] ?? null, + 'bootstrap' => $_SERVER['HTTP_X_MAINTENANCE_BOOTSTRAP'] ?? null]; +$status = 200; +$response = []; +if (($_SERVER['HTTP_AUTHORIZATION'] ?? '') !== 'Bearer fixture-worker') { + $status = 401; + $response = ['reason' => 'wrong_credential']; +} elseif ($path === '/api/cluster/info') { + $response = ['worker_protocol' => ['server_capabilities' => ['activity_lease_renewal' => true]]]; +} elseif ($path === '/api/worker/heartbeat') { + $response = ['worker_id' => 'owner', 'acknowledged' => true]; +} elseif (str_ends_with($path, '/activity-tasks/poll') && !isset($state['delivered'])) { + $state['delivered'] = true; + $state['expiry'] = $now + 3; + $state['progress_at'] = $now; + $state['business_at'] = isset($state['business_seconds']) ? $now + $state['business_seconds'] : null; + $deadlines = []; + if ($state['business_at'] !== null) { + $deadlines['start_to_close'] = $stamp($state['business_at']); + } + if (isset($state['heartbeat_seconds'])) { + $deadlines['heartbeat'] = $stamp($now + $state['heartbeat_seconds']); + } + $response = ['task' => ['task_id' => 'task', 'workflow_id' => 'workflow', 'run_id' => 'run', + 'activity_execution_id' => 'execution', 'activity_attempt_id' => 'attempt', 'attempt_number' => 1, + 'activity_type' => 'slow', 'task_queue' => 'queue', 'lease_owner' => 'owner', 'payload_codec' => 'avro', + 'namespace' => 'tenant', 'lease_expires_at' => $stamp($state['expiry']), 'deadlines' => $deadlines, + 'retry_policy' => ['heartbeat_timeout' => $state['heartbeat_seconds'] ?? null]], + 'server_time' => $stamp($now), 'poll_status' => 'leased']; +} elseif (str_ends_with($path, '/poll')) { + $response = ['task' => null, 'poll_status' => 'empty']; +} elseif (str_ends_with($path, '/renew-lease') || str_ends_with($path, '/heartbeat')) { + $heartbeat = str_ends_with($path, '/heartbeat'); + $live = $now < $state['expiry'] && !($state['closed'] ?? false) + && ($state['business_at'] === null || $now < $state['business_at']) + && (!isset($state['heartbeat_seconds']) || $now < $state['progress_at'] + $state['heartbeat_seconds']); + if ($live) { + $state['expiry'] = $now + 3; + if ($heartbeat) { + $state['progress_at'] = $now; + $state['heartbeats'] = ($state['heartbeats'] ?? 0) + 1; + } else { + $state['renewals'] = ($state['renewals'] ?? 0) + 1; + } + } + $response = ['task_id' => 'task', 'workflow_task_id' => 'task', 'workflow_instance_id' => 'workflow', + 'workflow_run_id' => 'run', 'activity_execution_id' => 'execution', 'activity_attempt_id' => 'attempt', + 'attempt_number' => 1, 'lease_owner' => 'owner', 'can_continue' => $live, 'cancel_requested' => false, + 'lease_renewed' => $live && !$heartbeat, 'heartbeat_recorded' => $live && $heartbeat, + 'lease_expires_at' => $stamp($state['expiry']), 'server_time' => $stamp($now), + 'last_heartbeat_at' => $stamp($state['progress_at']), + 'heartbeat_deadline_at' => isset($state['heartbeat_seconds']) + ? $stamp($state['progress_at'] + $state['heartbeat_seconds']) : null, + 'reason' => $live ? null : 'deadline_elapsed']; + if (($state['corrupt_after'] ?? PHP_INT_MAX) <= ($state['renewals'] ?? 0)) { + $response['activity_attempt_id'] = 'foreign-attempt'; + } + if (!$live) { + $status = 409; + } +} elseif (str_ends_with($path, '/complete') || str_ends_with($path, '/fail')) { + $live = $now < $state['expiry'] && ($state['business_at'] === null || $now < $state['business_at']); + if ($live) { + $state['outcomes'][] = $body; + $state['closed'] = true; + $response = ['recorded' => true, 'outcome' => str_ends_with($path, '/complete') ? 'completed' : 'failed']; + } else { + $status = 409; + $response = ['task_id' => 'task', 'activity_attempt_id' => 'attempt', 'reason' => 'stale_attempt', + 'recorded' => false, 'outcome' => 'failed']; + } +} else { + $status = 404; + $response = ['reason' => 'unexpected_fixture_route']; +} +rewind($handle); +ftruncate($handle, 0); +fwrite($handle, json_encode($state, JSON_THROW_ON_ERROR)); +flock($handle, LOCK_UN); +fclose($handle); +http_response_code($status); +header('Content-Type: application/json'); +echo json_encode($response, JSON_THROW_ON_ERROR); diff --git a/tests/Support/activity-lease-owner.php b/tests/Support/activity-lease-owner.php new file mode 100644 index 0000000..b115588 --- /dev/null +++ b/tests/Support/activity-lease-owner.php @@ -0,0 +1,13 @@ +pollActivityTaskResponse('owner', 'queue', 0); +$process = new DurableWorkflow\Worker\ActivityLeaseProcess($client, $poll['task'], 'owner', 'queue', + $poll['server_time'], hrtime(true) / 1e9, false, 10); +$handle = (new ReflectionProperty($process, 'process'))->getValue($process); +echo json_encode(['ready' => true, 'helper_pid' => proc_get_status($handle)['pid']], JSON_THROW_ON_ERROR)."\n"; +fflush(STDOUT); +sleep(20); diff --git a/tests/WorkerActivityLeaseMaintenanceTest.php b/tests/WorkerActivityLeaseMaintenanceTest.php new file mode 100644 index 0000000..61cf9c6 --- /dev/null +++ b/tests/WorkerActivityLeaseMaintenanceTest.php @@ -0,0 +1,277 @@ +directory = sys_get_temp_dir().'/dw-lease-'.bin2hex(random_bytes(8)); + mkdir($this->directory, 0700); + file_put_contents($this->directory.'/state.json', '{}'); + $socket = stream_socket_server('tcp://127.0.0.1:0'); + self::assertIsResource($socket); + $address = stream_socket_get_name($socket, false); + fclose($socket); + $this->uri = 'http://'.$address; + $this->server = proc_open([PHP_BINARY, '-S', $address, __DIR__.'/Support/activity-lease-http.php'], + [0 => ['file', '/dev/null', 'r'], 1 => ['file', '/dev/null', 'w'], 2 => ['file', '/dev/null', 'w']], + $pipes, env_vars: ['LEASE_TEST_STATE' => $this->directory.'/state.json']); + self::assertIsResource($this->server); + $until = microtime(true) + 5; + do { + $ready = @stream_socket_client('tcp://'.$address, $errno, $message, 0.1); + if (is_resource($ready)) { + fclose($ready); + return; + } + usleep(10000); + } while (microtime(true) < $until); + self::fail('The isolated HTTP fixture did not start.'); + } + + protected function tearDown(): void + { + if (is_resource($this->server)) { + proc_terminate($this->server); + proc_close($this->server); + } + if (isset($this->directory)) { + foreach (glob($this->directory.'/*') ?: [] as $path) { + unlink($path); + } + rmdir($this->directory); + } + } + + public function testBlockingCallbackStaysInTheOriginalProcessAndOutlastsTheRuntimeLease(): void + { + $worker = $this->worker(); + $captured = 0; + $parent = getmypid(); + $worker->registerActivity('slow', static function (ActivityContext $context) use (&$captured, $parent): int { + self::assertSame($parent, getmypid()); + usleep(4200000); + return ++$captured; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(1, $captured, 'Inline captured state must remain visible.'); + $state = $this->state(); + self::assertCount(1, $state['outcomes'] ?? [], 'A blocking callback must have one retained completion.'); + self::assertGreaterThanOrEqual(3, $state['renewals'] ?? 0); + self::assertSame(0, $state['heartbeats'] ?? 0, 'Runtime pings must not fabricate progress.'); + self::assertCount(1, $state['outcomes']); + $presence = array_values(array_filter($state['requests'], + static fn (array $request): bool => $request['path'] === '/api/worker/heartbeat')); + self::assertNotEmpty($presence); + self::assertSame(['workflow_available' => 0, 'activity_available' => 0, 'session_available' => 0], + $presence[0]['body']['task_slots']); + foreach ($state['requests'] as $request) { + self::assertSame('Bearer fixture-worker', $request['authorization']); + self::assertSame('tenant', $request['namespace']); + } + $renewals = $state['renewals']; + usleep(150000); + self::assertSame($renewals, $this->state()['renewals'], 'Joined maintenance must stop after execution.'); + } + + #[DataProvider('lateOutcomes')] + public function testExpiredBusinessAuthorityFencesBothReturnAndFailure(bool $throw): void + { + $worker = $this->worker(['business_seconds' => 2]); + $calls = 0; + $worker->registerActivity('slow', static function (ActivityContext $context) use ($throw, &$calls): string { + ++$calls; + usleep(2300000); + if ($throw) { + throw new \RuntimeException('Application error after authority expired.'); + } + return 'late'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(1, $calls); + self::assertSame([], $this->state()['outcomes'] ?? []); + self::assertSame([], array_values(array_filter($this->state()['requests'], + static fn (array $request): bool => str_ends_with($request['path'], '/complete') || str_ends_with($request['path'], '/fail')))); + } + + public static function lateOutcomes(): array + { + return [[false], [true]]; + } + + public function testRuntimePingsCannotReplaceRequiredApplicationHeartbeats(): void + { + $worker = $this->worker(['heartbeat_seconds' => 2]); + $worker->registerActivity('slow', static function (ActivityContext $context): string { + usleep(2300000); + return 'late'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(0, $this->state()['heartbeats'] ?? 0); + self::assertSame([], $this->state()['outcomes'] ?? []); + } + + public function testRealApplicationProgressIsRelayedWhileRuntimeMaintenanceContinues(): void + { + $worker = $this->worker(['heartbeat_seconds' => 3]); + $worker->registerActivity('slow', static function (ActivityContext $context): string { + for ($step = 0; $step < 4; ++$step) { + usleep(800000); + $context->heartbeat(['step' => $step]); + } + return 'done'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(4, $this->state()['heartbeats']); + self::assertCount(1, $this->state()['outcomes']); + } + + public function testChangedAttemptProofFencesOutcomeAfterCallbackReturns(): void + { + $worker = $this->worker(['corrupt_after' => 2]); + $calls = 0; + $worker->registerActivity('slow', static function (ActivityContext $context) use (&$calls): string { + ++$calls; + usleep(2300000); + return 'must not publish'; + }); + self::assertTrue($worker->tick(0)); + self::assertSame(1, $calls); + self::assertGreaterThanOrEqual(2, $this->state()['renewals']); + self::assertSame([], $this->state()['outcomes'] ?? []); + } + + public function testCustomHttpConfigurationMustNotSilentlyBecomeDefaultTransport(): void + { + $client = new Client($this->uri, transport: new Psr18Transport(new GuzzleClient(['verify' => '/private-ca.pem'])), + namespace: 'tenant', workerToken: 'fixture-worker'); + $this->expectException(\InvalidArgumentException::class); + $client->activityLeaseProcessConfiguration(); + } + + public function testMaintenanceStopsWhenTheOriginalOwnerIsKilled(): void + { + if (!function_exists('posix_kill')) { + self::markTestSkipped('This owner-death control uses POSIX process observation.'); + } + $autoload = dirname(__DIR__).'/vendor/autoload.php'; + $owner = proc_open([PHP_BINARY, __DIR__.'/Support/activity-lease-owner.php', $autoload, $this->uri], + [0 => ['file', '/dev/null', 'r'], 1 => ['pipe', 'w'], 2 => ['file', '/dev/null', 'w']], $pipes); + self::assertIsResource($owner); + $helper = null; + try { + stream_set_timeout($pipes[1], 6); + $ready = json_decode(fgets($pipes[1]) ?: '{}', true, 32, JSON_THROW_ON_ERROR); + self::assertTrue($ready['ready'] ?? false); + $helper = $ready['helper_pid']; + self::assertTrue(posix_kill($helper, 0)); + proc_terminate($owner, 9); + proc_close($owner); + $owner = null; + $until = microtime(true) + 3; + while (posix_kill($helper, 0) && microtime(true) < $until) { + usleep(10000); + } + self::assertFalse(posix_kill($helper, 0), 'Maintenance must not survive the original callback owner.'); + $count = $this->state()['renewals']; + usleep(1100000); + self::assertSame($count, $this->state()['renewals']); + self::assertSame([], $this->state()['outcomes'] ?? []); + } finally { + if (is_resource($pipes[1])) { + fclose($pipes[1]); + } + if (is_resource($owner)) { + proc_terminate($owner, 9); + proc_close($owner); + } + if (is_int($helper) && posix_kill($helper, 0)) { + posix_kill($helper, 9); + } + } + } + + public function testCustomBootstrapRecreatesItsHttpConfigurationInTheHelper(): void + { + $bootstrap = $this->directory.'/bootstrap.php'; + file_put_contents($bootstrap, 'uri, true).', ' + .'namespace: "tenant", workerToken: "fixture-worker", transport: new \\DurableWorkflow\\Transport\\Psr18Transport(' + .'new \\GuzzleHttp\\Client(["headers" => ["X-Maintenance-Bootstrap" => "retained"], "verify" => true])));'); + $client = (new Client($this->uri, namespace: 'tenant', workerToken: 'fixture-worker', + transport: new Psr18Transport(new GuzzleClient(['verify' => true]))))->withActivityLeaseBootstrap($bootstrap); + $worker = $this->worker(client: $client); + $worker->registerActivity('slow', static fn (ActivityContext $context): int => 1); + self::assertTrue($worker->tick(0)); + $renewals = array_values(array_filter($this->state()['requests'], + static fn (array $request): bool => str_ends_with($request['path'], '/renew-lease'))); + self::assertNotEmpty($renewals); + self::assertSame('retained', $renewals[0]['bootstrap']); + self::assertCount(1, $this->state()['outcomes']); + } + + public function testCallbackCannotPublishThroughItsCapturedClonedClientAfterExpiry(): void + { + $client = new Client($this->uri, namespace: 'tenant', workerToken: 'fixture-worker'); + $worker = $this->worker(['business_seconds' => 2], $client); + $refused = false; + $worker->registerActivity('slow', static function (ActivityContext $context) use ($client, &$refused): string { + usleep(2300000); + try { + $client->withNamespace('tenant')->completeActivityTask('task', 'attempt', 'owner', 'late'); + } catch (\DurableWorkflow\Worker\WorkflowClaimRevoked) { + $refused = true; + } + return 'also late'; + }); + self::assertTrue($worker->tick(0)); + self::assertTrue($refused); + self::assertSame([], $this->state()['outcomes'] ?? []); + self::assertSame([], array_values(array_filter($this->state()['requests'], + static fn (array $request): bool => str_ends_with($request['path'], '/complete')))); + } + + private function worker(array $configuration = [], ?Client $client = null): Worker + { + file_put_contents($this->directory.'/state.json', json_encode($configuration, JSON_THROW_ON_ERROR)); + $worker = new Worker($client ?? new Client($this->uri, namespace: 'tenant', + controlToken: 'must-not-reach-maintenance', workerToken: 'fixture-worker'), 'queue', workerId: 'owner'); + // tick() is the real polling/execution path; only registration setup is + // supplied directly so this fixture remains independent of rollout APIs. + (new \ReflectionProperty($worker, 'registered'))->setValue($worker, true); + (new \ReflectionProperty($worker, 'lastHeartbeatAt'))->setValue($worker, microtime(true)); + (new \ReflectionMethod($worker, 'captureRegistrationFence'))->invoke($worker, + ['server_capabilities' => ['activity_lease_renewal' => true]]); + + return $worker; + } + + private function state(): array + { + $file = fopen($this->directory.'/state.json', 'r'); + flock($file, LOCK_SH); + $state = json_decode(stream_get_contents($file), true, 32, JSON_THROW_ON_ERROR); + flock($file, LOCK_UN); + fclose($file); + + return $state; + } +} From 5c4a81c3e5bb39d055af0666b5a5ff8b45e3c8a7 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 10:29:39 +0000 Subject: [PATCH 4/6] Keep original process metrics and fence refused activity publications --- src/Client.php | 23 +++++++++++++++--- src/Worker.php | 2 +- src/Worker/ActivityLeaseMaintainer.php | 5 ++-- src/Worker/ActivityLeaseProcess.php | 25 ++++++++++++++++++-- tests/Support/activity-lease-http.php | 1 + tests/WorkerActivityLeaseMaintenanceTest.php | 20 ++++++++++++++++ 6 files changed, 68 insertions(+), 8 deletions(-) diff --git a/src/Client.php b/src/Client.php index 54ade9a..c0c98cf 100644 --- a/src/Client.php +++ b/src/Client.php @@ -1626,7 +1626,8 @@ public function closeWorkerSession(string $workerId, string $sessionId, string $ * @param array $taskSlots * @return array */ - public function heartbeatWorker(string $workerId, array $taskSlots = [], ?RequestBudget $budget = null): array + public function heartbeatWorker(string $workerId, array $taskSlots = [], ?RequestBudget $budget = null, + ?int $processId = null): array { if ($budget !== null && !$this->boundedWorkerRequests) { throw new \LogicException('A worker authority budget requires bounded worker requests.'); @@ -1635,7 +1636,7 @@ public function heartbeatWorker(string $workerId, array $taskSlots = [], ?Reques 'worker_id' => $workerId, 'task_slots' => $taskSlots ?: null, 'process_metrics' => [ - 'process_id' => getmypid(), + 'process_id' => $processId ?? getmypid(), 'process_uptime_seconds' => 0, ], ]), budget: $budget); @@ -2416,7 +2417,15 @@ private function worker(string $method, string $path, ?array $body = null, ?Requ $started = hrtime(true) / 1e9; $requestClient = $governed ? $this->withBoundedWorkerRequests() : $this; - $response = $requestClient->request($method, $path, true, $body, budget: $budget); + try { + $response = $requestClient->request($method, $path, true, $body, budget: $budget); + } catch (ServerException $error) { + if ($governed && str_ends_with($path, '/heartbeat') + && ($error->details['can_continue'] ?? null) === false) { + $lease->progress($error->details, $started, hrtime(true) / 1e9); + } + throw $error; + } if ($governed && str_ends_with($path, '/heartbeat')) { $lease->progress($response, $started, hrtime(true) / 1e9); } @@ -2606,6 +2615,14 @@ private function request( if ($body !== null) { $body = $this->payloadUploads->request($body, $method, $path, $worker, $headers, $budget); } + $activityLease = $worker ? $this->activityLeaseBinding->current() : null; + if ($activityLease !== null && $activityLease->governs($path)) { + // Uploading an external result is not publication authority. + // Check again before its final Worker POST and bound the send + // by the latest acknowledged grant, not an old cached lease. + $activityLease->checkRequest($path, $body); + $budget?->restrictAuthorityDeadline($activityLease->publicationDeadline()); + } if ($worker && $this->boundedWorkerRequests) { if (!$this->transport instanceof BoundedTransport || !$this->transport->supportsBoundedRequests()) { throw new InvalidArgumentException('The cooperative worker transport no longer supports bounded requests.'); diff --git a/src/Worker.php b/src/Worker.php index 6ca412c..098ef4f 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -1529,7 +1529,7 @@ private function workerPayloadProgress(string $path, ?array $body, array $respon $process = $this->activityLeaseProcess; - return static fn () => $process->check(); + return static fn () => $process->checkTransfer(); } $queryAuthority = null; if (!$activity) { diff --git a/src/Worker/ActivityLeaseMaintainer.php b/src/Worker/ActivityLeaseMaintainer.php index 5891c75..281c802 100644 --- a/src/Worker/ActivityLeaseMaintainer.php +++ b/src/Worker/ActivityLeaseMaintainer.php @@ -10,7 +10,7 @@ use DurableWorkflow\Transport\RequestBudget; use Throwable; -/** @internal Private metadata-only helper. Never loads or invokes an activity. */ +/** @internal Private metadata-only helper. Never dispatches an activity callback. */ final class ActivityLeaseMaintainer { private string $buffer = ''; @@ -160,7 +160,8 @@ private function requestRenewal(): array $identity = $this->authority->identity(); if (($this->start['registered'] ?? false) === true && hrtime(true) / 1e9 >= $this->nextPresence) { $ack = $this->client->heartbeatWorker((string) $identity['lease_owner'], - ['workflow_available' => 0, 'activity_available' => 0, 'session_available' => 0], $this->budget()); + ['workflow_available' => 0, 'activity_available' => 0, 'session_available' => 0], $this->budget(), + $this->start['parent_pid']); if (($ack['worker_id'] ?? null) !== $identity['lease_owner'] || ($ack['acknowledged'] ?? null) !== true) { throw new WorkflowClaimRevoked('activity_worker_unconfirmed', 'Original Worker presence was not acknowledged.'); } diff --git a/src/Worker/ActivityLeaseProcess.php b/src/Worker/ActivityLeaseProcess.php index 57f1282..6394a29 100644 --- a/src/Worker/ActivityLeaseProcess.php +++ b/src/Worker/ActivityLeaseProcess.php @@ -24,6 +24,7 @@ public static function available(): bool private string $buffer = ''; private int $sequence = 0; private int $parentPid = 0; + private float $lastTransferCheck = 0; private ?string $lostReason = null; private float $until; private readonly float $fixedUntil; @@ -55,7 +56,8 @@ public function __construct(Client $client, array $task, string $owner, string $ } $descriptors = [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['file', PHP_OS_FAMILY === 'Windows' ? 'NUL' : '/dev/null', 'w']]; - $this->process = proc_open([PHP_BINARY, __DIR__.'/activity-lease-maintainer.php', $autoload], + $this->process = proc_open([PHP_BINARY, '-d', 'auto_prepend_file=', '-d', 'auto_append_file=', + __DIR__.'/activity-lease-maintainer.php', $autoload], $descriptors, $this->pipes, options: ['bypass_shell' => true]); if (!is_resource($this->process)) { throw new \RuntimeException('Could not start independent activity lease maintenance.'); @@ -100,7 +102,26 @@ public function check(): void public function budget(): RequestBudget { - return new RequestBudget(30, min($this->until, $this->fixedUntil)); + return new RequestBudget(30, is_finite($this->fixedUntil) ? $this->fixedUntil : null); + } + + public function publicationDeadline(): float + { + return min($this->until, $this->fixedUntil); + } + + public function checkTransfer(): void + { + $now = hrtime(true) / 1e9; + if ($now >= $this->fixedUntil) { + $this->refuse('activity_execution_budget_expired'); + } + // Streaming checks bound staleness without an IPC round trip per chunk. + // Callback dispatch and terminal POSTs always obtain fresh proof. + if ($this->lostReason !== null || $now >= $this->until || $now - $this->lastTransferCheck >= 1) { + $this->check(); + $this->lastTransferCheck = $now; + } } /** @param array $reply */ diff --git a/tests/Support/activity-lease-http.php b/tests/Support/activity-lease-http.php index fb59f63..4023657 100644 --- a/tests/Support/activity-lease-http.php +++ b/tests/Support/activity-lease-http.php @@ -48,6 +48,7 @@ } elseif (str_ends_with($path, '/renew-lease') || str_ends_with($path, '/heartbeat')) { $heartbeat = str_ends_with($path, '/heartbeat'); $live = $now < $state['expiry'] && !($state['closed'] ?? false) + && !($heartbeat && ($state['deny_heartbeat'] ?? false)) && ($state['business_at'] === null || $now < $state['business_at']) && (!isset($state['heartbeat_seconds']) || $now < $state['progress_at'] + $state['heartbeat_seconds']); if ($live) { diff --git a/tests/WorkerActivityLeaseMaintenanceTest.php b/tests/WorkerActivityLeaseMaintenanceTest.php index 61cf9c6..4827981 100644 --- a/tests/WorkerActivityLeaseMaintenanceTest.php +++ b/tests/WorkerActivityLeaseMaintenanceTest.php @@ -84,6 +84,7 @@ public function testBlockingCallbackStaysInTheOriginalProcessAndOutlastsTheRunti self::assertNotEmpty($presence); self::assertSame(['workflow_available' => 0, 'activity_available' => 0, 'session_available' => 0], $presence[0]['body']['task_slots']); + self::assertSame($parent, $presence[0]['body']['process_metrics']['process_id']); foreach ($state['requests'] as $request) { self::assertSame('Bearer fixture-worker', $request['authorization']); self::assertSame('tenant', $request['namespace']); @@ -160,6 +161,25 @@ public function testChangedAttemptProofFencesOutcomeAfterCallbackReturns(): void self::assertSame([], $this->state()['outcomes'] ?? []); } + public function testCaughtHeartbeatHttpRefusalCannotRestoreCompletionAuthority(): void + { + $worker = $this->worker(['deny_heartbeat' => true]); + $caught = false; + $worker->registerActivity('slow', static function (ActivityContext $context) use (&$caught): string { + try { + $context->heartbeat(); + } catch (\Throwable) { + $caught = true; + } + return 'must not publish'; + }); + self::assertTrue($worker->tick(0)); + self::assertTrue($caught); + self::assertSame([], $this->state()['outcomes'] ?? []); + self::assertSame([], array_values(array_filter($this->state()['requests'], + static fn (array $request): bool => str_ends_with($request['path'], '/complete') || str_ends_with($request['path'], '/fail')))); + } + public function testCustomHttpConfigurationMustNotSilentlyBecomeDefaultTransport(): void { $client = new Client($this->uri, transport: new Psr18Transport(new GuzzleClient(['verify' => '/private-ca.pem'])), From cec79c1b1fc62633bf00ed4af023e6cc4817bcf2 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 12:04:08 +0000 Subject: [PATCH 5/6] Clarify activity maintenance authority and IPC stages --- src/Client.php | 11 +- src/Worker.php | 21 ++- src/Worker/ActivityLeaseAuthority.php | 138 +++++++++++------- src/Worker/ActivityLeaseMaintainer.php | 78 ++++++---- .../ActivityLeaseMaintenanceException.php | 10 ++ src/Worker/ActivityLeaseProcess.php | 113 ++++++++------ src/Worker/ActivityLeaseSession.php | 20 ++- src/Worker/activity-lease-maintainer.php | 2 +- tests/Support/activity-lease-owner.php | 2 +- 9 files changed, 243 insertions(+), 152 deletions(-) create mode 100644 src/Worker/ActivityLeaseMaintenanceException.php diff --git a/src/Client.php b/src/Client.php index c0c98cf..450d34a 100644 --- a/src/Client.php +++ b/src/Client.php @@ -67,6 +67,7 @@ final class Client implements WorkflowClientInterface private const WORKFLOW_TASK_WAITING_FOR_HISTORY_MESSAGE = 'Workflow task waiting for scheduled history.'; private const WORKFLOW_TASK_WAITING_FOR_HISTORY_TYPE = 'WorkflowTaskWaitingForHistory'; private const ACTIVITY_TASK_PATH = '/worker/activity-tasks/'; + private const CLUSTER_INFO_PATH = '/cluster/info'; private readonly string $baseUri; private readonly ?Authentication $authentication; @@ -273,7 +274,7 @@ public function health(): array public function clusterInfo(bool $includeDiagnostics = false): ClusterInfo { $this->observabilityContextSupported = false; - $path = $includeDiagnostics ? '/cluster/info?include=diagnostics' : '/cluster/info'; + $path = $includeDiagnostics ? self::CLUSTER_INFO_PATH.'?include=diagnostics' : self::CLUSTER_INFO_PATH; $info = $this->control('GET', $path); $this->observabilityContextSupported = TraceCarrier::supported($info); @@ -387,7 +388,7 @@ private function serviceCatalogRead(string $path, array $filters = [], ?string $ public function workerClusterInfo(): ClusterInfo { $this->observabilityContextSupported = false; - $info = $this->worker('GET', '/cluster/info'); + $info = $this->worker('GET', self::CLUSTER_INFO_PATH); $this->observabilityContextSupported = TraceCarrier::supported($info); return ClusterInfo::fromArray($info); @@ -2253,7 +2254,7 @@ public function renewActivityTaskLease( $budget->restrictRequestDuration(5); $budget->remainingSeconds(); $client = $this->withBoundedWorkerRequests(); - $info = $client->worker('GET', '/cluster/info', budget: $budget); + $info = $client->worker('GET', self::CLUSTER_INFO_PATH, budget: $budget); $protocol = $info['worker_protocol'] ?? null; $capabilities = is_array($protocol) ? ($protocol['server_capabilities'] ?? null) : null; if (!is_array($capabilities) || ($capabilities['activity_lease_renewal'] ?? null) !== true) { @@ -2399,7 +2400,7 @@ private function worker(string $method, string $path, ?array $body = null, ?Requ $lease = $this->activityLeaseBinding->current(); $governed = $lease !== null && $lease->governs($path); if ($governed) { - $lease->checkRequest($path, $body); + $lease->checkRequest($body); $budget ??= $lease->budget(); } if ($budget !== null && !$this->boundedWorkerRequests && !$governed) { @@ -2620,7 +2621,7 @@ private function request( // Uploading an external result is not publication authority. // Check again before its final Worker POST and bound the send // by the latest acknowledged grant, not an old cached lease. - $activityLease->checkRequest($path, $body); + $activityLease->checkRequest($body); $budget?->restrictAuthorityDeadline($activityLease->publicationDeadline()); } if ($worker && $this->boundedWorkerRequests) { diff --git a/src/Worker.php b/src/Worker.php index 098ef4f..2d4e0ac 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -16,6 +16,7 @@ use DurableWorkflow\Exception\TransportException; use DurableWorkflow\Worker\ActivityContext; use DurableWorkflow\Worker\ActivityLeaseProcess; +use DurableWorkflow\Worker\ActivityLeaseMaintenanceException; use DurableWorkflow\Worker\ActivityExecutionFailure; use DurableWorkflow\Worker\CapabilityManifest; use DurableWorkflow\Worker\CancellationHistory; @@ -385,12 +386,7 @@ public function run(int $pollTimeoutSeconds = 5): void } $this->registered = true; $this->lastHeartbeatAt = $this->now(); - if ($this->activityLeaseRenewalSupported && $this->activities !== []) { - if (!ActivityLeaseProcess::available()) { - throw new \RuntimeException('Automatic activity lease maintenance requires PHP CLI proc_open.'); - } - $this->client->activityLeaseProcessConfiguration(); - } + $this->prepareActivityLeaseMaintenance(); $this->diagnostic('worker.registered', [ 'worker_id' => $this->workerId, 'task_queue' => $this->taskQueue, @@ -1524,7 +1520,7 @@ private function workerPayloadProgress(string $path, ?array $body, array $respon } $this->activityLeaseProcess = new ActivityLeaseProcess($this->client, $task, $this->workerId, $this->taskQueue, $response['server_time'] ?? null, hrtime(true) / 1e9, - $this->registered, $this->heartbeatIntervalSeconds); + ['registered' => $this->registered, 'heartbeat_interval' => $this->heartbeatIntervalSeconds]); $this->unbindActivityLease = $this->client->bindActivityLease($this->activityLeaseProcess); $process = $this->activityLeaseProcess; @@ -3566,6 +3562,17 @@ private function heartbeatIfDue(?RequestBudget $budget = null): void $this->heartbeat($budget); } + private function prepareActivityLeaseMaintenance(): void + { + if (!$this->activityLeaseRenewalSupported || $this->activities === []) { + return; + } + if (!ActivityLeaseProcess::available()) { + throw new ActivityLeaseMaintenanceException('Automatic activity lease maintenance requires PHP CLI proc_open.'); + } + $this->client->activityLeaseProcessConfiguration(); + } + private function stopActivityLease(): void { if ($this->unbindActivityLease !== null) { diff --git a/src/Worker/ActivityLeaseAuthority.php b/src/Worker/ActivityLeaseAuthority.php index 9f952b7..0924ae4 100644 --- a/src/Worker/ActivityLeaseAuthority.php +++ b/src/Worker/ActivityLeaseAuthority.php @@ -43,49 +43,11 @@ public function __construct( ) { $this->clock = $clock ?? static fn (): float => hrtime(true) / 1e9; try { - foreach (['task_id', 'workflow_id', 'run_id', 'activity_execution_id', 'activity_attempt_id', - 'lease_owner', 'task_queue'] as $key) { - if (!is_string($task[$key] ?? null) || trim($task[$key]) === '' || strlen($task[$key]) > 255) { - throw new \UnexpectedValueException('The delivery lacks its original '.$key.'.'); - } - } - if ($task['lease_owner'] !== $owner || $task['task_queue'] !== $queue - || ($task['namespace'] ?? $namespace) !== $namespace - || !is_int($task['attempt_number'] ?? null) || $task['attempt_number'] < 1 - || !is_finite($capturedAt) || $capturedAt < 0 || $capturedAt > ($this->clock)()) { - throw new \UnexpectedValueException('The delivery does not belong to this Worker.'); - } - $this->identity = ['task_id' => $task['task_id'], 'workflow_task_id' => $task['task_id'], - 'workflow_instance_id' => $task['workflow_id'], 'workflow_run_id' => $task['run_id'], - 'activity_execution_id' => $task['activity_execution_id'], - 'activity_attempt_id' => $task['activity_attempt_id'], 'attempt_number' => $task['attempt_number'], - 'lease_owner' => $owner]; + $this->identity = $this->originalIdentity($task, $namespace, $owner, $queue); $this->serverTime = self::timestamp($serverTime); $this->leaseExpiry = self::timestamp($task['lease_expires_at'] ?? null); $this->leaseUntil = $capturedAt + self::duration($this->leaseExpiry, $this->serverTime); - $deadlines = $task['deadlines'] ?? []; - if (!is_array($deadlines)) { - throw new \UnexpectedValueException('Invalid activity deadlines.'); - } - $bounds = []; - foreach (['start_to_close', 'schedule_to_close', 'run', 'execution'] as $key) { - if (isset($deadlines[$key])) { - $bounds[] = $capturedAt + self::timestamp($deadlines[$key]) - $this->serverTime; - } - } - $this->businessUntil = $bounds === [] ? null : min($bounds); - $policy = $task['retry_policy'] ?? null; - $interval = is_array($policy) ? ($policy['heartbeat_timeout'] ?? null) : null; - $this->heartbeatInterval = is_int($interval) && $interval > 0 ? $interval : null; - $this->heartbeatDeadline = isset($deadlines['heartbeat']) ? self::timestamp($deadlines['heartbeat']) : null; - if ($this->heartbeatDeadline !== null && ($this->heartbeatInterval === null - || $this->heartbeatDeadline - $this->serverTime > $this->heartbeatInterval)) { - throw new \UnexpectedValueException('Missing original heartbeat policy.'); - } - $this->heartbeatUntil = $this->heartbeatDeadline === null ? null - : $capturedAt + $this->heartbeatDeadline - $this->serverTime; - $this->heartbeatTime = $this->heartbeatDeadline === null ? null - : $this->heartbeatDeadline - $this->heartbeatInterval; + $this->captureDeadlines($task); $this->session = isset($task['worker_session']) ? new ActivityLeaseSession($task['worker_session'], $namespace, $owner, $this->serverTime, $capturedAt) : null; @@ -95,6 +57,59 @@ public function __construct( $this->remaining(); } + /** @param array $task + * @return array + */ + private function originalIdentity(array $task, string $namespace, string $owner, string $queue): array + { + foreach (['task_id', 'workflow_id', 'run_id', 'activity_execution_id', 'activity_attempt_id', + 'lease_owner', 'task_queue'] as $key) { + if (!is_string($task[$key] ?? null) || trim($task[$key]) === '' || strlen($task[$key]) > 255) { + throw new \UnexpectedValueException('The delivery lacks its original '.$key.'.'); + } + } + if ($task['lease_owner'] !== $owner || $task['task_queue'] !== $queue + || ($task['namespace'] ?? $namespace) !== $namespace + || !is_int($task['attempt_number'] ?? null) || $task['attempt_number'] < 1 + || !is_finite($this->capturedAt) || $this->capturedAt < 0 || $this->capturedAt > ($this->clock)()) { + throw new \UnexpectedValueException('The delivery does not belong to this Worker.'); + } + + return ['task_id' => $task['task_id'], 'workflow_task_id' => $task['task_id'], + 'workflow_instance_id' => $task['workflow_id'], 'workflow_run_id' => $task['run_id'], + 'activity_execution_id' => $task['activity_execution_id'], + 'activity_attempt_id' => $task['activity_attempt_id'], 'attempt_number' => $task['attempt_number'], + 'lease_owner' => $owner]; + } + + /** @param array $task */ + private function captureDeadlines(array $task): void + { + $deadlines = $task['deadlines'] ?? []; + if (!is_array($deadlines)) { + throw new \UnexpectedValueException('Invalid activity deadlines.'); + } + $bounds = []; + foreach (['start_to_close', 'schedule_to_close', 'run', 'execution'] as $key) { + if (isset($deadlines[$key])) { + $bounds[] = $this->capturedAt + self::timestamp($deadlines[$key]) - $this->serverTime; + } + } + $this->businessUntil = $bounds === [] ? null : min($bounds); + $policy = $task['retry_policy'] ?? null; + $interval = is_array($policy) ? ($policy['heartbeat_timeout'] ?? null) : null; + $this->heartbeatInterval = is_int($interval) && $interval > 0 ? $interval : null; + $this->heartbeatDeadline = isset($deadlines['heartbeat']) ? self::timestamp($deadlines['heartbeat']) : null; + if ($this->heartbeatDeadline !== null && ($this->heartbeatInterval === null + || $this->heartbeatDeadline - $this->serverTime > $this->heartbeatInterval)) { + throw new \UnexpectedValueException('Missing original heartbeat policy.'); + } + $this->heartbeatUntil = $this->heartbeatDeadline === null ? null + : $this->capturedAt + $this->heartbeatDeadline - $this->serverTime; + $this->heartbeatTime = $this->heartbeatDeadline === null ? null + : $this->heartbeatDeadline - $this->heartbeatInterval; + } + public function remaining(): float { return $this->remainingAt(($this->clock)()); @@ -128,15 +143,7 @@ public function acknowledge(array $reply, float $started): void $this->remainingAt($received); try { $this->assertRoundTrip($started, $received); - foreach ($this->identity as $key => $value) { - if (($reply[$key] ?? null) !== $value) { - throw new \UnexpectedValueException('Activity renewal changed '.$key.'.'); - } - } - if (($reply['lease_renewed'] ?? null) !== true || ($reply['heartbeat_recorded'] ?? null) !== false) { - throw new \UnexpectedValueException('Runtime renewal did not acknowledge a metadata-only grant.'); - } - $this->assertContinuation($reply); + $this->assertRenewalIdentity($reply); $server = self::timestamp($reply['server_time'] ?? null); $expiry = self::timestamp($reply['lease_expires_at'] ?? null); $until = $started + self::duration($expiry, $server); @@ -159,12 +166,7 @@ public function acknowledge(array $reply, float $started): void throw new \UnexpectedValueException('Replayed activity renewal.'); } $this->observeProgress($reply, $server, $started); - if ($this->session !== null) { - if (!is_array($reply['worker_session'] ?? null)) { - throw new \UnexpectedValueException('Required session grant is missing from renewal.'); - } - $this->session->acknowledge($reply['worker_session'], $server, $started); - } + $this->acknowledgeRequiredSession($reply, $server, $started); $this->leaseUntil = max($this->leaseUntil, $until); $this->serverTime = $server; $this->leaseExpiry = $expiry; @@ -175,6 +177,32 @@ public function acknowledge(array $reply, float $started): void $this->remaining(); } + /** @param array $reply */ + private function assertRenewalIdentity(array $reply): void + { + foreach ($this->identity as $key => $value) { + if (($reply[$key] ?? null) !== $value) { + throw new \UnexpectedValueException('Activity renewal changed '.$key.'.'); + } + } + if (($reply['lease_renewed'] ?? null) !== true || ($reply['heartbeat_recorded'] ?? null) !== false) { + throw new \UnexpectedValueException('Runtime renewal did not acknowledge a metadata-only grant.'); + } + $this->assertContinuation($reply); + } + + /** @param array $reply */ + private function acknowledgeRequiredSession(array $reply, float $server, float $started): void + { + if ($this->session === null) { + return; + } + if (!is_array($reply['worker_session'] ?? null)) { + throw new \UnexpectedValueException('Required session grant is missing from renewal.'); + } + $this->session->acknowledge($reply['worker_session'], $server, $started); + } + /** * Only an actual user heartbeat can move the recorded progress deadline. * The receipt instant is retained when the parent relays it over private IPC. diff --git a/src/Worker/ActivityLeaseMaintainer.php b/src/Worker/ActivityLeaseMaintainer.php index 281c802..bd4f1d0 100644 --- a/src/Worker/ActivityLeaseMaintainer.php +++ b/src/Worker/ActivityLeaseMaintainer.php @@ -103,37 +103,13 @@ private static function client(array $configuration): Client private function loop(): void { while (!feof(STDIN) && $this->parentPresent()) { - $now = hrtime(true) / 1e9; - $reply = null; - $started = $now; - if ($this->lostReason === null && $now >= $this->nextRenewal) { - try { - $reply = $this->requestRenewal(); - } catch (Throwable $error) { - $this->recordMaintenanceError($error); - } - } + $started = hrtime(true) / 1e9; + $reply = $this->renewWhenDue($started); // Actual parent progress can overtake the in-flight runtime reply. // Drain those receipts before evaluating the old heartbeat ceiling. $commands = $this->readCommands(); - foreach ($commands as $command) { - if (($command['operation'] ?? null) === 'progress' && $this->lostReason === null) { - try { - $this->authority->acknowledgeProgress($command['reply'], - $command['started'], $command['received']); - } catch (Throwable $error) { - $this->recordMaintenanceError($error); - } - } - } - if ($reply !== null && $this->lostReason === null) { - try { - $this->authority->acknowledge($reply, $started); - $this->scheduleRenewal(); - } catch (Throwable $error) { - $this->recordMaintenanceError($error); - } - } + $this->applyProgressCommands($commands); + $this->applyRenewal($reply, $started); foreach ($commands as $command) { if (!in_array($command['operation'] ?? null, ['check', 'progress'], true)) { $this->lostReason = 'activity_maintenance_invalid_command'; @@ -144,6 +120,50 @@ private function loop(): void } } + /** @return array|null */ + private function renewWhenDue(float $started): ?array + { + if ($this->lostReason !== null || $started < $this->nextRenewal) { + return null; + } + try { + return $this->requestRenewal(); + } catch (Throwable $error) { + $this->recordMaintenanceError($error); + + return null; + } + } + + /** @param list> $commands */ + private function applyProgressCommands(array $commands): void + { + foreach ($commands as $command) { + if (($command['operation'] ?? null) !== 'progress' || $this->lostReason !== null) { + continue; + } + try { + $this->authority->acknowledgeProgress($command['reply'], $command['started'], $command['received']); + } catch (Throwable $error) { + $this->recordMaintenanceError($error); + } + } + } + + /** @param array|null $reply */ + private function applyRenewal(?array $reply, float $started): void + { + if ($reply === null || $this->lostReason !== null) { + return; + } + try { + $this->authority->acknowledge($reply, $started); + $this->scheduleRenewal(); + } catch (Throwable $error) { + $this->recordMaintenanceError($error); + } + } + private function maintainInitialGrant(): void { $started = hrtime(true) / 1e9; @@ -243,7 +263,7 @@ private static function write(array $message): void { $line = json_encode($message, JSON_THROW_ON_ERROR)."\n"; if (fwrite(STDOUT, $line) !== strlen($line)) { - throw new \RuntimeException('Original activity process is unavailable.'); + throw new ActivityLeaseMaintenanceException('Original activity process is unavailable.'); } fflush(STDOUT); } diff --git a/src/Worker/ActivityLeaseMaintenanceException.php b/src/Worker/ActivityLeaseMaintenanceException.php new file mode 100644 index 0000000..535ad56 --- /dev/null +++ b/src/Worker/ActivityLeaseMaintenanceException.php @@ -0,0 +1,10 @@ + */ private readonly array $identity; - /** @param array $task */ + /** @param array $task + * @param array{registered: bool, heartbeat_interval: int} $presence + */ public function __construct(Client $client, array $task, string $owner, string $queue, - mixed $serverTime, float $captured, bool $registered, int $heartbeatInterval) + mixed $serverTime, float $captured, array $presence) { $this->parentPid = getmypid(); $authority = new ActivityLeaseAuthority($task, $client->namespace, $owner, $queue, $serverTime, $captured); @@ -41,7 +43,7 @@ public function __construct(Client $client, array $task, string $owner, string $ $this->until = hrtime(true) / 1e9 + $authority->remaining(); $this->fixedUntil = $authority->fixedBound(); if (!self::available()) { - throw new \RuntimeException('Automatic activity lease maintenance requires PHP CLI proc_open.'); + throw new ActivityLeaseMaintenanceException('Automatic activity lease maintenance requires PHP CLI proc_open.'); } $configuration = $client->activityLeaseProcessConfiguration(); $autoload = null; @@ -52,7 +54,7 @@ public function __construct(Client $client, array $task, string $owner, string $ } } if ($autoload === null) { - throw new \RuntimeException('Activity maintenance requires this SDK\'s Composer autoloader.'); + throw new ActivityLeaseMaintenanceException('Activity maintenance requires this SDK\'s Composer autoloader.'); } $descriptors = [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['file', PHP_OS_FAMILY === 'Windows' ? 'NUL' : '/dev/null', 'w']]; @@ -60,7 +62,7 @@ public function __construct(Client $client, array $task, string $owner, string $ __DIR__.'/activity-lease-maintainer.php', $autoload], $descriptors, $this->pipes, options: ['bypass_shell' => true]); if (!is_resource($this->process)) { - throw new \RuntimeException('Could not start independent activity lease maintenance.'); + throw new ActivityLeaseMaintenanceException('Could not start independent activity lease maintenance.'); } foreach ($this->pipes as $pipe) { stream_set_blocking($pipe, false); @@ -71,7 +73,7 @@ public function __construct(Client $client, array $task, string $owner, string $ 'attempt_number', 'lease_owner', 'task_queue', 'namespace', 'lease_expires_at', 'deadlines', 'retry_policy', 'worker_session'])), 'server_time' => $serverTime, 'captured_at' => $captured, 'owner' => $owner, 'queue' => $queue, - 'parent_pid' => getmypid(), 'registered' => $registered, 'heartbeat_interval' => $heartbeatInterval]); + 'parent_pid' => getmypid()] + $presence); } catch (Throwable $error) { $this->stop(); throw $error; @@ -86,7 +88,7 @@ public function governs(string $path): bool } /** @param array|null $body */ - public function checkRequest(string $path, ?array $body): void + public function checkRequest(?array $body): void { if (($body['activity_attempt_id'] ?? null) !== $this->identity['activity_attempt_id'] || ($body['lease_owner'] ?? null) !== $this->identity['lease_owner']) { @@ -146,17 +148,11 @@ public function stop(): void return; } if (is_resource($this->process)) { - $until = hrtime(true) / 1e9 + 0.25; - while ($this->running() && hrtime(true) / 1e9 < $until) { - usleep(5000); - } + $this->waitForStop(); if ($this->running()) { // This stops maintenance only. It is never callback-stop proof. proc_terminate($this->process); - $until = hrtime(true) / 1e9 + 0.25; - while ($this->running() && hrtime(true) / 1e9 < $until) { - usleep(5000); - } + $this->waitForStop(); if ($this->running()) { proc_terminate($this->process, 9); } @@ -166,6 +162,14 @@ public function stop(): void $this->process = null; } + private function waitForStop(): void + { + $until = hrtime(true) / 1e9 + 0.25; + while ($this->running() && hrtime(true) / 1e9 < $until) { + usleep(5000); + } + } + /** @phpstan-impure Process status changes independently of this PHP process. */ private function running(): bool { @@ -200,38 +204,8 @@ private function exchange(string $operation, array $body = []): void $this->refuse('activity_maintenance_message_too_large'); } try { - $offset = 0; - while ($offset < strlen($line)) { - $written = fwrite($this->pipes[0], substr($line, $offset)); - if ($written === false || hrtime(true) / 1e9 >= $deadline) { - $this->refuse('activity_maintenance_unavailable'); - } - $offset += $written; - if ($written === 0) { - usleep(1000); - } - } - while (!str_contains($this->buffer, "\n")) { - $part = stream_get_contents($this->pipes[1], 65537 - strlen($this->buffer)); - if ($part === false || feof($this->pipes[1]) || strlen($this->buffer) + strlen($part) > 65536 - || hrtime(true) / 1e9 >= $deadline) { - $this->refuse('activity_maintenance_unavailable'); - } - $this->buffer .= $part; - usleep(1000); - } - [$raw, $this->buffer] = explode("\n", $this->buffer, 2); - $reply = json_decode($raw, true, 32, JSON_THROW_ON_ERROR); - $now = hrtime(true) / 1e9; - if (!is_array($reply) || ($reply['id'] ?? null) !== $id || ($reply['can_continue'] ?? null) !== true - || !is_numeric($reply['until'] ?? null) || !is_numeric($reply['monotonic'] ?? null) - || $reply['monotonic'] < $started || $reply['monotonic'] > $now - || $reply['until'] <= $now || $now >= $deadline) { - $reason = is_array($reply) && is_string($reply['reason'] ?? null) - ? $reply['reason'] : 'activity_maintenance_unconfirmed'; - $this->refuse($reason); - } - $this->until = (float) $reply['until']; + $this->writeMessage($line, $deadline); + $this->acceptReply($this->readReply($deadline), $id, $started, $deadline); } catch (WorkflowClaimRevoked $error) { throw $error; } catch (Throwable $error) { @@ -240,6 +214,51 @@ private function exchange(string $operation, array $body = []): void } } + private function writeMessage(string $line, float $deadline): void + { + $offset = 0; + while ($offset < strlen($line)) { + $written = fwrite($this->pipes[0], substr($line, $offset)); + if ($written === false || hrtime(true) / 1e9 >= $deadline) { + $this->refuse('activity_maintenance_unavailable'); + } + $offset += $written; + if ($written === 0) { + usleep(1000); + } + } + } + + private function readReply(float $deadline): mixed + { + while (!str_contains($this->buffer, "\n")) { + $part = stream_get_contents($this->pipes[1], 65537 - strlen($this->buffer)); + if ($part === false || feof($this->pipes[1]) || strlen($this->buffer) + strlen($part) > 65536 + || hrtime(true) / 1e9 >= $deadline) { + $this->refuse('activity_maintenance_unavailable'); + } + $this->buffer .= $part; + usleep(1000); + } + [$raw, $this->buffer] = explode("\n", $this->buffer, 2); + + return json_decode($raw, true, 32, JSON_THROW_ON_ERROR); + } + + private function acceptReply(mixed $reply, int $id, float $started, float $deadline): void + { + $now = hrtime(true) / 1e9; + if (!is_array($reply) || ($reply['id'] ?? null) !== $id || ($reply['can_continue'] ?? null) !== true + || !is_numeric($reply['until'] ?? null) || !is_numeric($reply['monotonic'] ?? null) + || $reply['monotonic'] < $started || $reply['monotonic'] > $now + || $reply['until'] <= $now || $now >= $deadline) { + $reason = is_array($reply) && is_string($reply['reason'] ?? null) + ? $reply['reason'] : 'activity_maintenance_unconfirmed'; + $this->refuse($reason); + } + $this->until = (float) $reply['until']; + } + private function refuse(string $reason): never { $this->lostReason = $reason; diff --git a/src/Worker/ActivityLeaseSession.php b/src/Worker/ActivityLeaseSession.php index 6e1163a..280ffe0 100644 --- a/src/Worker/ActivityLeaseSession.php +++ b/src/Worker/ActivityLeaseSession.php @@ -87,6 +87,19 @@ private static function options(array $value): array || !is_bool($value['create_if_missing']) || !is_bool($value['allow_reacquire_after_failure'])) { throw new \UnexpectedValueException('Malformed original session options.'); } + self::validateDurationsAndRequirements($value); + + $options = []; + foreach ([...$required, 'queue', 'requirements'] as $key) { + $options[$key] = $value[$key] ?? null; + } + + return $options; + } + + /** @param array $value */ + private static function validateDurationsAndRequirements(array $value): void + { foreach (['lease_seconds', 'ttl_seconds', 'max_concurrent_activities'] as $key) { if (!is_int($value[$key]) || $value[$key] < 1) { throw new \UnexpectedValueException('Invalid session duration or concurrency.'); @@ -97,12 +110,5 @@ private static function options(array $value): array throw new \UnexpectedValueException('Invalid session requirement.'); } } - - $options = []; - foreach ([...$required, 'queue', 'requirements'] as $key) { - $options[$key] = $value[$key] ?? null; - } - - return $options; } } diff --git a/src/Worker/activity-lease-maintainer.php b/src/Worker/activity-lease-maintainer.php index 7c82c21..fe21a87 100644 --- a/src/Worker/activity-lease-maintainer.php +++ b/src/Worker/activity-lease-maintainer.php @@ -6,5 +6,5 @@ if (PHP_SAPI !== 'cli' || !isset($argv[1]) || !is_file($argv[1])) { exit(1); } -require $argv[1]; +require_once $argv[1]; DurableWorkflow\Worker\ActivityLeaseMaintainer::run(); diff --git a/tests/Support/activity-lease-owner.php b/tests/Support/activity-lease-owner.php index b115588..7ee8429 100644 --- a/tests/Support/activity-lease-owner.php +++ b/tests/Support/activity-lease-owner.php @@ -6,7 +6,7 @@ $client = new DurableWorkflow\Client($argv[2], namespace: 'tenant', workerToken: 'fixture-worker'); $poll = $client->pollActivityTaskResponse('owner', 'queue', 0); $process = new DurableWorkflow\Worker\ActivityLeaseProcess($client, $poll['task'], 'owner', 'queue', - $poll['server_time'], hrtime(true) / 1e9, false, 10); + $poll['server_time'], hrtime(true) / 1e9, ['registered' => false, 'heartbeat_interval' => 10]); $handle = (new ReflectionProperty($process, 'process'))->getValue($process); echo json_encode(['ready' => true, 'helper_pid' => proc_get_status($handle)['pid']], JSON_THROW_ON_ERROR)."\n"; fflush(STDOUT); From e13d84b95c142cb6feaa962255bde3e30923c86e Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 14:10:09 +0000 Subject: [PATCH 6/6] test: preserve private CA and rotating auth during activity maintenance --- tests/Support/activity-lease-http.php | 3 +- tests/Support/activity-lease-tls.php | 62 ++++++++++++++++ tests/WorkerActivityLeaseMaintenanceTest.php | 76 ++++++++++++++++++++ 3 files changed, 140 insertions(+), 1 deletion(-) create mode 100644 tests/Support/activity-lease-tls.php diff --git a/tests/Support/activity-lease-http.php b/tests/Support/activity-lease-http.php index 4023657..68e4107 100644 --- a/tests/Support/activity-lease-http.php +++ b/tests/Support/activity-lease-http.php @@ -18,7 +18,8 @@ 'bootstrap' => $_SERVER['HTTP_X_MAINTENANCE_BOOTSTRAP'] ?? null]; $status = 200; $response = []; -if (($_SERVER['HTTP_AUTHORIZATION'] ?? '') !== 'Bearer fixture-worker') { +$expectedToken = isset($state['authentication_file']) ? trim(file_get_contents($state['authentication_file'])) : 'fixture-worker'; +if (($_SERVER['HTTP_AUTHORIZATION'] ?? '') !== 'Bearer '.$expectedToken) { $status = 401; $response = ['reason' => 'wrong_credential']; } elseif ($path === '/api/cluster/info') { diff --git a/tests/Support/activity-lease-tls.php b/tests/Support/activity-lease-tls.php new file mode 100644 index 0000000..bdf418f --- /dev/null +++ b/tests/Support/activity-lease-tls.php @@ -0,0 +1,62 @@ + ['local_cert' => $certificate, 'local_pk' => $key]]); +$server = stream_socket_server('tcp://127.0.0.1:0', $errno, $error, STREAM_SERVER_BIND | STREAM_SERVER_LISTEN, $context); +if ($server === false) { + exit(1); +} +file_put_contents($ready, stream_socket_get_name($server, false)); +$until = microtime(true) + 30; +while (microtime(true) < $until) { + $client = @stream_socket_accept($server, 1); + if ($client === false) { + continue; + } + stream_set_timeout($client, 3); + if (@stream_socket_enable_crypto($client, true, STREAM_CRYPTO_METHOD_TLS_SERVER) !== true) { + fclose($client); + continue; + } + $peer = @stream_socket_client($upstream, $errno, $error, 3); + if ($peer === false) { + fclose($client); + continue; + } + stream_set_timeout($peer, 3); + $headers = ''; + $length = 0; + while (($line = fgets($client, 65537)) !== false) { + $headers .= $line; + if (strlen($headers) > 65536) { + break; + } + if (preg_match('/^Content-Length:\s*(\d+)\s*$/i', trim($line), $matches)) { + $length = (int) $matches[1]; + } + if ($line === "\r\n") { + break; + } + } + if (strlen($headers) <= 65536 && $length <= 65536 && str_ends_with($headers, "\r\n\r\n")) { + $body = ''; + while (strlen($body) < $length) { + $chunk = fread($client, $length - strlen($body)); + if ($chunk === false || $chunk === '') { + break; + } + $body .= $chunk; + } + if (strlen($body) === $length) { + fwrite($peer, $headers.$body); + stream_copy_to_stream($peer, $client); + } + } + fclose($peer); + fclose($client); +} +fclose($server); diff --git a/tests/WorkerActivityLeaseMaintenanceTest.php b/tests/WorkerActivityLeaseMaintenanceTest.php index 4827981..a7bc519 100644 --- a/tests/WorkerActivityLeaseMaintenanceTest.php +++ b/tests/WorkerActivityLeaseMaintenanceTest.php @@ -16,6 +16,8 @@ final class WorkerActivityLeaseMaintenanceTest extends TestCase { /** @var resource|null */ private $server; + /** @var resource|null */ + private $tlsServer = null; private string $directory; private string $uri; @@ -50,6 +52,10 @@ protected function setUp(): void protected function tearDown(): void { + if (is_resource($this->tlsServer)) { + proc_terminate($this->tlsServer); + proc_close($this->tlsServer); + } if (is_resource($this->server)) { proc_terminate($this->server); proc_close($this->server); @@ -248,6 +254,76 @@ public function testCustomBootstrapRecreatesItsHttpConfigurationInTheHelper(): v self::assertCount(1, $this->state()['outcomes']); } + public function testPrivateCaAndRotatingAuthenticationSurviveTheIndependentBootstrap(): void + { + if (!extension_loaded('openssl')) { + self::markTestSkipped('The actual private-CA control requires OpenSSL.'); + } + $httpUri = $this->uri; + $this->uri = $this->privateTlsEndpoint($httpUri); + try { + (new Client($this->uri, workerToken: 'fixture-worker'))->workerClusterInfo(); + self::fail('The generated private certificate must not be trusted by default.'); + } catch (\DurableWorkflow\Exception\ServerException $error) { + self::assertInstanceOf(\DurableWorkflow\Exception\TransportException::class, $error->getPrevious()); + self::assertStringContainsString('certificate', strtolower($error->getMessage())); + } + self::assertSame([], $this->state()['requests'] ?? [], 'Untrusted TLS must not reach the HTTP fixture.'); + + $token = $this->directory.'/worker-token'; + file_put_contents($token, 'fixture-worker'); + $bootstrap = $this->directory.'/tls-bootstrap.php'; + file_put_contents($bootstrap, 'uri, true).', ' + .'authentication: new class implements \\DurableWorkflow\\Auth\\Authentication {' + .'public function headers(bool $workerRequest): array { return ["Authorization" => "Bearer ".trim(file_get_contents(' + .var_export($token, true).'))]; } }, namespace: "tenant", ' + .'transport: new \\DurableWorkflow\\Transport\\Psr18Transport(new \\GuzzleHttp\\Client(["verify" => ' + .var_export($this->directory.'/certificate.pem', true).'])));'); + $client = (require $bootstrap)->withActivityLeaseBootstrap($bootstrap); + $worker = $this->worker(['authentication_file' => $token], $client); + $worker->registerActivity('slow', static function (ActivityContext $context) use ($token): int { + file_put_contents($token, 'rotated-fixture-worker'); + usleep(4200000); + return 17; + }); + self::assertTrue($worker->tick(0)); + self::assertCount(1, $this->state()['outcomes'] ?? []); + $renewals = array_values(array_filter($this->state()['requests'], + static fn (array $request): bool => str_ends_with($request['path'], '/renew-lease'))); + self::assertGreaterThanOrEqual(3, count($renewals)); + self::assertSame('Bearer fixture-worker', $renewals[0]['authorization']); + self::assertSame('Bearer rotated-fixture-worker', $renewals[count($renewals) - 1]['authorization']); + } + + private function privateTlsEndpoint(string $upstream): string + { + $config = $this->directory.'/openssl.cnf'; + file_put_contents($config, "[req]\ndistinguished_name=dn\n[dn]\n[v3]\nsubjectAltName=IP:127.0.0.1\nbasicConstraints=critical,CA:TRUE\nextendedKeyUsage=serverAuth\n"); + $key = openssl_pkey_new(['private_key_bits' => 2048]); + self::assertNotFalse($key); + $request = openssl_csr_new(['commonName' => '127.0.0.1'], $key, ['config' => $config, 'digest_alg' => 'sha256']); + self::assertNotFalse($request); + $certificate = openssl_csr_sign($request, null, $key, 1, + ['config' => $config, 'x509_extensions' => 'v3', 'digest_alg' => 'sha256']); + self::assertNotFalse($certificate); + self::assertTrue(openssl_x509_export_to_file($certificate, $this->directory.'/certificate.pem')); + self::assertTrue(openssl_pkey_export_to_file($key, $this->directory.'/key.pem')); + $ready = $this->directory.'/tls-ready'; + $this->tlsServer = proc_open([PHP_BINARY, __DIR__.'/Support/activity-lease-tls.php', + $this->directory.'/certificate.pem', $this->directory.'/key.pem', + str_replace('http://', 'tcp://', $upstream), $ready], + [0 => ['file', '/dev/null', 'r'], 1 => ['file', '/dev/null', 'w'], 2 => ['file', '/dev/null', 'w']], $pipes); + self::assertIsResource($this->tlsServer); + $until = microtime(true) + 5; + do { + if (is_file($ready)) { + return 'https://'.file_get_contents($ready); + } + usleep(10000); + } while (microtime(true) < $until); + self::fail('The isolated TLS fixture did not start.'); + } + public function testCallbackCannotPublishThroughItsCapturedClonedClientAfterExpiry(): void { $client = new Client($this->uri, namespace: 'tenant', workerToken: 'fixture-worker');