diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ec5213eb..0a7de500 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -3,7 +3,7 @@ name: CI on: pull_request: push: - branches: [main] + branches: [main, 3.9.x] workflow_dispatch: inputs: cooperative_qualification: diff --git a/.github/workflows/sonar.yml b/.github/workflows/sonar.yml index 8aad74ca..f473c859 100644 --- a/.github/workflows/sonar.yml +++ b/.github/workflows/sonar.yml @@ -2,9 +2,9 @@ name: SonarQube on: push: - branches: [main] + branches: [main, 3.9.x] pull_request: - branches: [main] + branches: [main, 3.9.x] permissions: contents: read diff --git a/examples/laravel-http/composer.lock b/examples/laravel-http/composer.lock index fa277aae..73da04a7 100644 --- a/examples/laravel-http/composer.lock +++ b/examples/laravel-http/composer.lock @@ -5310,16 +5310,16 @@ }, { "name": "symfony/http-client", - "version": "v7.4.21", + "version": "v7.4.23", "source": { "type": "git", "url": "https://github.com/symfony/http-client.git", - "reference": "121bf3a973bfd852db85ec32d0d64e9fbed4449b" + "reference": "ed012bfca0f7b45d97016efcdf9b0639c52a1722" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/http-client/zipball/121bf3a973bfd852db85ec32d0d64e9fbed4449b", - "reference": "121bf3a973bfd852db85ec32d0d64e9fbed4449b", + "url": "https://api.github.com/repos/symfony/http-client/zipball/ed012bfca0f7b45d97016efcdf9b0639c52a1722", + "reference": "ed012bfca0f7b45d97016efcdf9b0639c52a1722", "shasum": "" }, "require": { @@ -5387,7 +5387,7 @@ "http" ], "support": { - "source": "https://github.com/symfony/http-client/tree/v7.4.21" + "source": "https://github.com/symfony/http-client/tree/v7.4.23" }, "funding": [ { @@ -5407,7 +5407,7 @@ "type": "tidelift" } ], - "time": "2026-10-13T15:31:38+00:00" + "time": "2026-11-04T08:27:22+00:00" }, { "name": "symfony/http-client-contracts", @@ -5493,16 +5493,16 @@ }, { "name": "symfony/http-foundation", - "version": "v7.4.21", + "version": "v7.4.23", "source": { "type": "git", "url": "https://github.com/symfony/http-foundation.git", - "reference": "58c89144c7a36dd89c3d01917e9fa3a72d8aaf31" + "reference": "7fb73cffa06e2bcc2cb523fcccb1e8174106b206" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/http-foundation/zipball/58c89144c7a36dd89c3d01917e9fa3a72d8aaf31", - "reference": "58c89144c7a36dd89c3d01917e9fa3a72d8aaf31", + "url": "https://api.github.com/repos/symfony/http-foundation/zipball/7fb73cffa06e2bcc2cb523fcccb1e8174106b206", + "reference": "7fb73cffa06e2bcc2cb523fcccb1e8174106b206", "shasum": "" }, "require": { @@ -5551,7 +5551,7 @@ "description": "Defines an object-oriented layer for the HTTP specification", "homepage": "https://symfony.com", "support": { - "source": "https://github.com/symfony/http-foundation/tree/v7.4.21" + "source": "https://github.com/symfony/http-foundation/tree/v7.4.23" }, "funding": [ { @@ -5571,20 +5571,20 @@ "type": "tidelift" } ], - "time": "2026-10-08T16:56:12+00:00" + "time": "2026-11-04T08:27:22+00:00" }, { "name": "symfony/http-kernel", - "version": "v7.4.21", + "version": "v7.4.23", "source": { "type": "git", "url": "https://github.com/symfony/http-kernel.git", - "reference": "4779fce7dedacb8801b2b61c703ef99159502331" + "reference": "94c57f4c5c45a311d244a0aeadb72b14ee41d25d" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/http-kernel/zipball/4779fce7dedacb8801b2b61c703ef99159502331", - "reference": "4779fce7dedacb8801b2b61c703ef99159502331", + "url": "https://api.github.com/repos/symfony/http-kernel/zipball/94c57f4c5c45a311d244a0aeadb72b14ee41d25d", + "reference": "94c57f4c5c45a311d244a0aeadb72b14ee41d25d", "shasum": "" }, "require": { @@ -5670,7 +5670,7 @@ "description": "Provides a structured process for converting a Request into a Response", "homepage": "https://symfony.com", "support": { - "source": "https://github.com/symfony/http-kernel/tree/v7.4.21" + "source": "https://github.com/symfony/http-kernel/tree/v7.4.23" }, "funding": [ { @@ -5690,7 +5690,7 @@ "type": "tidelift" } ], - "time": "2026-10-14T06:44:08+00:00" + "time": "2026-11-04T11:26:18+00:00" }, { "name": "symfony/mailer", @@ -6932,16 +6932,16 @@ }, { "name": "symfony/routing", - "version": "v7.4.21", + "version": "v7.4.23", "source": { "type": "git", "url": "https://github.com/symfony/routing.git", - "reference": "7a344cf871532d081875bfdd18f3dff8dca36c5b" + "reference": "0000ade604b7bc798cb1e75edf6f17df48185f73" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/routing/zipball/7a344cf871532d081875bfdd18f3dff8dca36c5b", - "reference": "7a344cf871532d081875bfdd18f3dff8dca36c5b", + "url": "https://api.github.com/repos/symfony/routing/zipball/0000ade604b7bc798cb1e75edf6f17df48185f73", + "reference": "0000ade604b7bc798cb1e75edf6f17df48185f73", "shasum": "" }, "require": { @@ -6993,7 +6993,7 @@ "url" ], "support": { - "source": "https://github.com/symfony/routing/tree/v7.4.21" + "source": "https://github.com/symfony/routing/tree/v7.4.23" }, "funding": [ { @@ -7013,7 +7013,7 @@ "type": "tidelift" } ], - "time": "2026-10-11T18:40:59+00:00" + "time": "2026-11-04T08:27:22+00:00" }, { "name": "symfony/service-contracts", @@ -7542,16 +7542,16 @@ }, { "name": "symfony/yaml", - "version": "v7.4.21", + "version": "v7.4.23", "source": { "type": "git", "url": "https://github.com/symfony/yaml.git", - "reference": "044907929c7999814378be22da34722c7a86bad4" + "reference": "1e601207ffe627a2e621faa9d661f1d32dcf1cec" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/symfony/yaml/zipball/044907929c7999814378be22da34722c7a86bad4", - "reference": "044907929c7999814378be22da34722c7a86bad4", + "url": "https://api.github.com/repos/symfony/yaml/zipball/1e601207ffe627a2e621faa9d661f1d32dcf1cec", + "reference": "1e601207ffe627a2e621faa9d661f1d32dcf1cec", "shasum": "" }, "require": { @@ -7594,7 +7594,7 @@ "description": "Loads and dumps YAML files", "homepage": "https://symfony.com", "support": { - "source": "https://github.com/symfony/yaml/tree/v7.4.21" + "source": "https://github.com/symfony/yaml/tree/v7.4.23" }, "funding": [ { @@ -7614,7 +7614,7 @@ "type": "tidelift" } ], - "time": "2026-10-12T12:34:30+00:00" + "time": "2026-11-04T08:27:22+00:00" }, { "name": "tijsverkoyen/css-to-inline-styles", diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index 0fef9a49..f3cbb8e0 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -127,6 +127,17 @@ "content_patterns": [ "function\\s+workflowTaskHistory\\s*\\(" ] + }, + { + "glob": "src/Worker/WorkflowPollGrant.php" + }, + { + "glob": "src/Worker.php", + "content_patterns": [ + "function\\s+confirmWorkflowPollGrant\\s*\\(", + "function\\s+observePoll\\s*\\(", + "function\\s+workerPayloadProgress\\s*\\(" + ] } ] } diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php index 60b7ca8e..59fa26f5 100644 --- a/scripts/ci/run-replay-regression-fixture.php +++ b/scripts/ci/run-replay-regression-fixture.php @@ -84,6 +84,7 @@ final class ReplayRegressionTransport implements Transport private ?array $activityCompletionBody = null; private bool $activityCompleted = false; private bool $workflowClaimRepaired = false; + private int $workflowRenewals = 0; /** * @param list> $tasks @@ -112,7 +113,13 @@ public function send(string $method, string $uri, array $headers, ?array $body = $task['workflow_task_attempt'] = 2; } - return ['task' => $task, 'poll_status' => 'leased']; + $reply = ['task' => $task, 'poll_status' => 'leased']; + if ($task['workflow_type'] === 'golden.workflow-fresh-grant') { + $reply['server_time'] = '2026-07-18T00:00:00.000000Z'; + $reply['task']['lease_expires_at'] = '2026-07-18T00:00:30.000000Z'; + } + + return $reply; } $task = $this->taskFor($uri, 'history'); @@ -130,6 +137,7 @@ public function send(string $method, string $uri, array $headers, ?array $body = $task = $this->taskFor($uri, 'heartbeat'); if ($task !== null) { + ++$this->workflowRenewals; if ($this->workflowLeaseRecovery && !$this->workflowClaimRepaired) { $this->workflowClaimRepaired = true; --$this->workflowPoll; @@ -315,6 +323,11 @@ public function completedCommands(): array return $this->completedCommands; } + public function workflowRenewals(): int + { + return $this->workflowRenewals; + } + /** @return array|null */ private function taskFor(string $uri, string $operation): ?array { @@ -609,6 +622,9 @@ private static function executeLayout( } $observed = ['command_sequence' => $commands]; + if ($workflowType === 'golden.workflow-fresh-grant') { + $observed['initial_workflow_renewals'] = $transport->workflowRenewals(); + } if (count($commands) === 1) { $observed = array_merge($observed, $commands[0]); } @@ -804,6 +820,7 @@ private static function workflow(string $workflowType): callable return 'not-cancelled'; }, 'golden.worker-update' => static fn (WorkflowContext $context): array => [], + 'golden.workflow-fresh-grant' => static fn (WorkflowContext $context): string => 'fresh-grant', 'golden.workflow-stream' => static function (WorkflowContext $context): string { $context->appendWorkflowStream('tokens', [ new WorkflowStreamAppendItem(['token' => 'hello']), diff --git a/src/Worker.php b/src/Worker.php index 1c6086e3..33ef5199 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -44,6 +44,7 @@ use DurableWorkflow\Worker\WorkflowClaimDeferred; use DurableWorkflow\Worker\WorkflowClaimRevoked; use DurableWorkflow\Worker\WorkflowContext; +use DurableWorkflow\Worker\WorkflowPollGrant; use DurableWorkflow\Worker\WorkflowEntryArguments; use DurableWorkflow\Worker\StickyWorkflowCache; use DurableWorkflow\Worker\WorkerSession; @@ -112,6 +113,9 @@ final class Worker private ?QueryTaskAuthority $queryTaskAuthority = null; private bool $activityLeaseRenewalSupported = false; private ?ActivityLeaseProcess $activityLeaseProcess = null; + private ?ActivityLeaseProcess $activityLeaseHelper = null; + private ?WorkflowPollGrant $workflowPollGrant = null; + private ?float $workflowPollStartedAt = null; /** @var (\Closure(): void)|null */ private ?\Closure $unbindActivityLease = null; private ?CancellationRequest $claimCancellation = null; @@ -354,6 +358,9 @@ public function requestShutdown(): void return; } $this->shutdownRequested = true; + if ($this->activityLeaseProcess === null) { + $this->activityLeaseHelper?->stop(); + } $this->diagnostic('worker.shutdown_requested', ['worker_id' => $this->workerId]); } @@ -419,6 +426,8 @@ public function run(int $pollTimeoutSeconds = 5): void throw $exception; } finally { $this->stopActivityLease(); + $this->activityLeaseHelper?->stop(); + $this->activityLeaseHelper = null; $this->closeWorkerSessions(); $this->stickyCache->clear(); if ($this->registered) { @@ -1141,6 +1150,10 @@ private function transientRetryDelay(int $attempt, mixed $retryAfterSeconds): fl */ private function observePoll(string $taskKind, \Closure $operation): array { + if ($taskKind === 'workflow') { + $this->workflowPollGrant = null; + $this->workflowPollStartedAt = hrtime(true) / 1e9; + } if (!$this->metricsObserver->enabled()) { return $operation(); } @@ -1404,7 +1417,7 @@ private function executeWorkflowTask(array $task): void $this->observeClaimCancellation($task['cancellation_request']); } $history = $this->completeHistory($task, $leaseOwner, $attempt); - if (!$this->renewWorkflowTaskLease($taskId, $leaseOwner, $attempt)) { + if (!$this->confirmWorkflowPollGrant($task, $taskId, $leaseOwner, $attempt)) { return; } if ($this->client->observabilityContextAvailable()) { @@ -1511,6 +1524,7 @@ function (Throwable $failure) use ($taskId, $leaseOwner, $attempt): void { $attempt, ); } finally { + $this->workflowPollGrant = null; $traceScope?->__invoke(); $this->claimCancellation = null; $this->claimDeliveredCancellationId = null; @@ -1534,6 +1548,19 @@ private function isTimedOutWorkflowCompletion(array $task, ServerException $exce && ($details['workflow_task_attempt'] ?? null) === (int) ($task['workflow_task_attempt'] ?? 1); } + /** @param array $task */ + private function confirmWorkflowPollGrant(array $task, string $taskId, string $owner, int $attempt): bool + { + $grant = $this->workflowPollGrant; + $this->workflowPollGrant = null; + if ($this->shutdownRequested) { + return false; + } + + return ($grant?->isFreshFor($task, hrtime(true) / 1e9) ?? false) + || $this->renewWorkflowTaskLease($taskId, $owner, $attempt); + } + private function renewWorkflowTaskLease(string $taskId, string $leaseOwner, int $taskAttempt): bool { $retryAttempt = 0; @@ -1595,9 +1622,16 @@ private function workerPayloadProgress(string $path, ?array $body, array $respon 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, - ['registered' => $this->registered, 'heartbeat_interval' => $this->heartbeatIntervalSeconds]); + $presence = ['registered' => $this->registered, 'heartbeat_interval' => $this->heartbeatIntervalSeconds, + 'presence_due_in' => max(0.0, $this->heartbeatIntervalSeconds - $this->elapsedSinceHeartbeat())]; + if ($this->activityLeaseHelper === null) { + $this->activityLeaseHelper = new ActivityLeaseProcess($this->client, $task, $this->workerId, + $this->taskQueue, $response['server_time'] ?? null, hrtime(true) / 1e9, $presence); + } else { + $this->activityLeaseHelper->start($this->client, $task, $this->workerId, + $this->taskQueue, $response['server_time'] ?? null, hrtime(true) / 1e9, $presence); + } + $this->activityLeaseProcess = $this->activityLeaseHelper; $this->unbindActivityLease = $this->client->bindActivityLease($this->activityLeaseProcess); $process = $this->activityLeaseProcess; @@ -1658,6 +1692,9 @@ private function workerPayloadProgress(string $path, ?array $body, array $respon }; } if ($path === '/worker/workflow-tasks/poll') { + // Capture metadata before external payload loading and history paging. + $this->workflowPollGrant = $this->workflowPollStartedAt === null ? null + : WorkflowPollGrant::observe($response, $this->workerId, $this->workflowPollStartedAt, hrtime(true) / 1e9); $task = $response['task'] ?? null; if (!is_array($task)) { return null; @@ -3674,8 +3711,11 @@ private function stopActivityLease(): void ($this->unbindActivityLease)(); $this->unbindActivityLease = null; } - $this->activityLeaseProcess?->stop(); + $this->activityLeaseProcess?->finish(); $this->activityLeaseProcess = null; + if ($this->shutdownRequested) { + $this->activityLeaseHelper?->stop(); + } } private function heartbeat(?RequestBudget $budget = null): void diff --git a/src/Worker/ActivityLeaseMaintainer.php b/src/Worker/ActivityLeaseMaintainer.php index a947e77a..b1a0b191 100644 --- a/src/Worker/ActivityLeaseMaintainer.php +++ b/src/Worker/ActivityLeaseMaintainer.php @@ -21,7 +21,7 @@ final class ActivityLeaseMaintainer private readonly Client $client; /** @param array $start */ - private function __construct(private readonly array $start) + private function __construct(private readonly array $start, ?Client $client = null) { $configuration = $start['client']; if (!is_array($configuration) || !is_array($start['task'] ?? null) @@ -29,34 +29,76 @@ private function __construct(private readonly array $start) || !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->client = $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']); + $due = $start['presence_due_in'] ?? 0; + $this->nextPresence = (float) $start['captured_at'] + + (is_int($due) || is_float($due) ? max(0, min(10, $due)) : 0); } public static function run(): void { - $first = fgets(STDIN, 65537); - $start = []; - try { - if ($first === false || !str_ends_with($first, "\n")) { + stream_set_blocking(STDIN, false); + $parent = null; + $configuration = null; + $client = null; + $proven = false; + while (true) { + $start = []; + try { + $start = self::awaitStart($parent); + $key = json_encode($start['client'] ?? null, JSON_THROW_ON_ERROR); + if ($start === null || ($start['operation'] ?? null) !== 'start' + || ($configuration !== null && $key !== $configuration) + || ($parent !== null && ($start['parent_pid'] ?? null) !== $parent)) { + return; + } + $maintainer = new self($start, $client); + $parent = $start['parent_pid']; + $configuration = $key; + $client = $maintainer->client; + $maintainer->maintainInitialGrant($proven); + $proven = true; + $maintainer->respond($start); + if (!$maintainer->loop()) { + return; + } + } catch (Throwable) { + // Bootstrap exceptions can contain application secrets. Only a + // bounded protocol reason is ever written to the parent pipe. + self::write(['id' => $start['id'] ?? 0, + 'can_continue' => false, 'reason' => 'activity_maintenance_bootstrap_failed']); return; } - $start = json_decode($first, true, 32, JSON_THROW_ON_ERROR); - if (!is_array($start) || ($start['operation'] ?? null) !== 'start') { - return; + } + } + + /** @return array|null */ + private static function awaitStart(?int $parent): ?array + { + $buffer = ''; + while (!str_contains($buffer, "\n")) { + if ($parent !== null && function_exists('posix_getppid') && posix_getppid() !== $parent) { + return null; + } + $read = [STDIN]; + $write = $except = null; + if (@stream_select($read, $write, $except, 1) === false) { + // Windows process pipes are not selectable; keep the existing + // bounded nonblocking-read path on that platform and on EINTR. + usleep(10000); } - $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']); + $part = stream_get_contents(STDIN, 65537 - strlen($buffer)); + if ($part === false || strlen($part) + strlen($buffer) > 65536 || feof(STDIN)) { + return null; + } + $buffer .= $part; } + [$line, $remaining] = explode("\n", $buffer, 2); + $command = $remaining === '' ? json_decode($line, true, 32, JSON_THROW_ON_ERROR) : null; + + return is_array($command) ? $command : null; } /** @param array $configuration */ @@ -111,7 +153,7 @@ private static function client(array $configuration): Client workerRequestTimeoutSeconds: $workerRequestTimeout === null ? null : (float) $workerRequestTimeout); } - private function loop(): void + private function loop(): bool { while (!feof(STDIN) && $this->parentPresent()) { $started = hrtime(true) / 1e9; @@ -121,20 +163,43 @@ private function loop(): void $commands = $this->readCommands(); $this->applyProgressCommands($commands); $this->applyRenewal($reply, $started); - foreach ($commands as $command) { + foreach ($commands as $index => $command) { + if (($command['operation'] ?? null) === 'finish') { + if ($index !== array_key_last($commands) || $this->buffer !== '') { + return false; + } + self::write(['id' => $command['id'] ?? 0, 'finished' => true]); + return true; + } if (!in_array($command['operation'] ?? null, ['check', 'progress'], true)) { $this->lostReason = 'activity_maintenance_invalid_command'; } $this->respond($command); } - usleep(10000); + $this->waitForCommand(); + } + + return false; + } + + private function waitForCommand(): void + { + $due = ($this->start['registered'] ?? false) === true + ? min($this->nextRenewal, $this->nextPresence) : $this->nextRenewal; + $delay = $this->lostReason === null ? max(0, min(1, $due - hrtime(true) / 1e9)) : 1; + $read = [STDIN]; + $write = $except = null; + if (@stream_select($read, $write, $except, (int) $delay, (int) (($delay - (int) $delay) * 1e6)) === false) { + usleep((int) min(10000, $delay * 1e6)); } } /** @return array|null */ private function renewWhenDue(float $started): ?array { - if ($this->lostReason !== null || $started < $this->nextRenewal) { + $due = ($this->start['registered'] ?? false) === true + ? min($this->nextRenewal, $this->nextPresence) : $this->nextRenewal; + if ($this->lostReason !== null || $started < $due) { return null; } try { @@ -175,8 +240,13 @@ private function applyRenewal(?array $reply, float $started): void } } - private function maintainInitialGrant(): void + private function maintainInitialGrant(bool $proven): void { + $age = hrtime(true) / 1e9 - (float) $this->start['captured_at']; + if ($proven && $age >= 0 && $age < 1 && $this->authority->remaining() >= 3) { + $this->scheduleRenewal(); + return; + } $started = hrtime(true) / 1e9; $this->authority->acknowledge($this->requestRenewal(), $started); $this->scheduleRenewal(); @@ -220,6 +290,7 @@ private function recordMaintenanceError(Throwable $error): void || $error->isTransientUpstreamFailure() || $error->isStorageAdmissionFailure())) { // Unavailability adds no time. Retry only within the preceding grant. $this->nextRenewal = hrtime(true) / 1e9 + 0.1; + $this->nextPresence = max($this->nextPresence, $this->nextRenewal); return; } diff --git a/src/Worker/ActivityLeaseProcess.php b/src/Worker/ActivityLeaseProcess.php index 84ff265e..cd3d4906 100644 --- a/src/Worker/ActivityLeaseProcess.php +++ b/src/Worker/ActivityLeaseProcess.php @@ -9,7 +9,7 @@ use DurableWorkflow\Transport\RequestBudget; use Throwable; -/** @internal A fresh PHP process maintains leases while the original callback stays inline. */ +/** @internal One Worker's reusable helper. Activity authority is replaced only after confirmed finish. */ final class ActivityLeaseProcess { public static function available(): bool @@ -26,18 +26,32 @@ public static function available(): bool private int $parentPid = 0; private float $lastTransferCheck = 0; private ?string $lostReason = null; - private float $until; - private readonly float $fixedUntil; + private float $until = 0; + private float $fixedUntil = 0; + private bool $active = false; + private string $configurationKey = ''; /** @var array */ - private readonly array $identity; + private array $identity = []; /** @param array $task - * @param array{registered: bool, heartbeat_interval: int} $presence + * @param array{registered: bool, heartbeat_interval: int, presence_due_in?: float} $presence */ public function __construct(Client $client, array $task, string $owner, string $queue, mixed $serverTime, float $captured, array $presence) { $this->parentPid = getmypid(); + $this->start($client, $task, $owner, $queue, $serverTime, $captured, $presence); + } + + /** @param array $task + * @param array{registered: bool, heartbeat_interval: int, presence_due_in?: float} $presence + */ + public function start(Client $client, array $task, string $owner, string $queue, + mixed $serverTime, float $captured, array $presence): void + { + if ($this->parentPid !== getmypid() || $this->active) { + throw new ActivityLeaseMaintenanceException('Only the original idle Worker can start activity maintenance.'); + } $authority = new ActivityLeaseAuthority($task, $client->namespace, $owner, $queue, $serverTime, $captured); $this->identity = $authority->identity(); $this->until = hrtime(true) / 1e9 + $authority->remaining(); @@ -56,17 +70,23 @@ public function __construct(Client $client, array $task, string $owner, string $ if ($autoload === null) { 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']]; - $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 ActivityLeaseMaintenanceException('Could not start independent activity lease maintenance.'); - } - foreach ($this->pipes as $pipe) { - stream_set_blocking($pipe, false); + $key = hash('sha256', json_encode([$configuration, $autoload], JSON_THROW_ON_ERROR)); + if (!$this->running() || $key !== $this->configurationKey) { + $this->stop(); + $descriptors = [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], + 2 => ['file', PHP_OS_FAMILY === 'Windows' ? 'NUL' : '/dev/null', 'w']]; + $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 ActivityLeaseMaintenanceException('Could not start independent activity lease maintenance.'); + } + array_walk($this->pipes, static fn ($pipe): bool => stream_set_blocking($pipe, false)); + $this->configurationKey = $key; } + $this->lostReason = null; + $this->lastTransferCheck = 0; + $this->active = true; 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', @@ -82,7 +102,7 @@ public function __construct(Client $client, array $task, string $owner, string $ public function governs(string $path): bool { - return in_array($path, array_map(fn (string $action): string => '/worker/activity-tasks/' + return $this->active && 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); } @@ -134,6 +154,8 @@ public function progress(array $reply, float $started, float $received): void public function stop(): void { + $this->active = false; + $this->buffer = ''; foreach ($this->pipes as $pipe) { if (is_resource($pipe)) { fclose($pipe); @@ -162,6 +184,32 @@ public function stop(): void $this->process = null; } + /** Stop this claim; retain the process only after an unambiguous idle acknowledgment. */ + public function finish(): void + { + if (!$this->active) { + return; + } + if ($this->parentPid !== getmypid() || $this->lostReason !== null + || $this->buffer !== '' || !$this->running()) { + $this->stop(); + return; + } + try { + $id = ++$this->sequence; + $deadline = hrtime(true) / 1e9 + 1; + $this->writeMessage(json_encode(['id' => $id, 'operation' => 'finish'], JSON_THROW_ON_ERROR)."\n", $deadline); + $reply = $this->readReply($deadline); + if (!is_array($reply) || ($reply['id'] ?? null) !== $id + || ($reply['finished'] ?? null) !== true || $this->buffer !== '') { + $this->stop(); + } + } catch (Throwable) { + $this->stop(); + } + $this->active = false; + } + private function waitForStop(): void { $until = hrtime(true) / 1e9 + 0.25; @@ -188,6 +236,9 @@ private function exchange(string $operation, array $body = []): void throw new WorkflowClaimRevoked('activity_supervisor_process_mismatch', 'Only the original Worker process can use its activity maintenance channel.'); } + if (!$this->active) { + $this->refuse('activity_maintenance_idle'); + } if ($this->lostReason !== null) { $this->refuse($this->lostReason); } diff --git a/src/Worker/WorkflowPollGrant.php b/src/Worker/WorkflowPollGrant.php new file mode 100644 index 00000000..6ec7d3db --- /dev/null +++ b/src/Worker/WorkflowPollGrant.php @@ -0,0 +1,72 @@ + $identity */ + private function __construct( + private readonly array $identity, + private readonly float $receivedAt, + private readonly float $until, + ) { + } + + /** @param array $response */ + public static function observe(array $response, string $workerId, float $startedAt, float $receivedAt): ?self + { + $task = $response['task'] ?? null; + if (!is_array($task) || ($response['poll_status'] ?? null) !== 'leased' + || !is_string($task['task_id'] ?? null) || $task['task_id'] === '' + || ($task['lease_owner'] ?? null) !== $workerId + || !is_int($task['workflow_task_attempt'] ?? null) || $task['workflow_task_attempt'] < 1 + || !is_finite($startedAt) || !is_finite($receivedAt) || $receivedAt < $startedAt) { + return null; + } + $server = self::timestamp($response['server_time'] ?? null); + $expiry = self::timestamp($task['lease_expires_at'] ?? null); + if ($server === null || $expiry === null) { + return null; + } + + // Charge the whole request, including transit, without trusting the worker's wall clock. + return new self(array_intersect_key($task, array_flip([ + 'task_id', 'lease_owner', 'workflow_task_attempt', 'lease_expires_at', + ])), $receivedAt, $startedAt + ($expiry - $server)); + } + + /** @param array $task */ + public function isFreshFor(array $task, float $now): bool + { + if (!is_finite($now) || $now < $this->receivedAt || $now - $this->receivedAt >= 1 + || $this->until - $now < 10) { + return false; + } + foreach ($this->identity as $key => $value) { + if (($task[$key] ?? null) !== $value) { + return false; + } + } + + return true; + } + + private 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) { + return null; + } + $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) { + return null; + } + + return (float) $date->format('U.u'); + } +} diff --git a/tests/Support/ReplayRegressionFixture.php b/tests/Support/ReplayRegressionFixture.php index f1619fa3..2c8b21e2 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -82,6 +82,10 @@ public static function execute(array $fixture): array throw new RuntimeException("{$identity}.history must not be empty."); } + if ($workflowType === 'golden.workflow-fresh-grant') { + return self::executeFreshWorkflowPoll($identity, $input, $history, $fixture); + } + $workflowTasks = $fixture['workflow_tasks'] ?? null; if ($workflowTasks !== null) { if (!is_array($workflowTasks) || !array_is_list($workflowTasks) || $workflowTasks === []) { @@ -801,6 +805,62 @@ private static function executeLocalActivityContext(string $identity, array $inp return $observed; } + /** + * @param list $input + * @param list> $history + * @param array $fixture + * @return list> + */ + private static function executeFreshWorkflowPoll(string $identity, array $input, array $history, array $fixture): array + { + $codec = new AvroPayloadCodec(); + $fence = ['task_id' => 'fresh-task', 'workflow_task_attempt' => 1, 'lease_owner' => 'fresh-worker']; + $commands = null; + $renewals = 0; + $transport = new FakeTransport(handler: static function (string $method, string $uri, array $headers, ?array $body) use ( + $input, $history, $codec, $fence, &$commands, &$renewals, + ): array { + if (str_ends_with($uri, '/workflow-tasks/poll')) { + return ['poll_status' => 'leased', 'server_time' => '2026-07-18T00:00:00.000000Z', + 'task' => $fence + ['workflow_id' => 'fresh-workflow', 'run_id' => 'fresh-run', + 'workflow_type' => 'golden.workflow-fresh-grant', 'payload_codec' => 'avro', + 'arguments' => $codec->envelope($input), 'history_events' => $history, + 'lease_expires_at' => '2026-07-18T00:00:30.000000Z']]; + } + if (str_ends_with($uri, '/fresh-task/heartbeat')) { + ++$renewals; + + return $fence + ['renewed' => true]; + } + if (str_ends_with($uri, '/fresh-task/complete')) { + if (($body['lease_owner'] ?? null) !== $fence['lease_owner'] + || ($body['workflow_task_attempt'] ?? null) !== $fence['workflow_task_attempt']) { + throw new RuntimeException('Fresh poll completion changed its original fence.'); + } + $commands = $body['commands'] ?? null; + + return ['completed' => true]; + } + if (str_ends_with($uri, '/poll')) { + return ['task' => null, 'poll_status' => 'empty']; + } + + throw new RuntimeException('Unexpected fresh workflow poll request: '.$method.' '.$uri); + }); + $worker = new Worker(new Client('https://replay.invalid', transport: $transport, codec: $codec), + 'regression-corpus', workerId: 'fresh-worker'); + $worker->registerWorkflow('golden.workflow-fresh-grant', static fn (WorkflowContext $context): string => 'fresh-grant'); + $worker->tick(0); + if (!is_array($commands)) { + throw new RuntimeException('Fresh poll did not complete its workflow task.'); + } + $commands = array_map(static fn (array $command): array => self::decodeEnvelopes($command, $codec), $commands); + $observed = ['command_sequence' => $commands, 'initial_workflow_renewals' => $renewals] + $commands[0]; + self::assertMatches($fixture['expected'], $observed, "{$identity}.expected"); + + return $commands; + } + /** @return array */ private static function taskAttributes(string $workflowType): array { diff --git a/tests/Support/activity-lease-http.php b/tests/Support/activity-lease-http.php index 88a53af2..c0d14556 100644 --- a/tests/Support/activity-lease-http.php +++ b/tests/Support/activity-lease-http.php @@ -26,9 +26,15 @@ $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'])) { +} elseif (str_ends_with($path, '/activity-tasks/poll') + && ($state['delivery_count'] ?? 0) < ($state['activity_count'] ?? 1)) { $state['delivered'] = true; - $state['expiry'] = $now + 3; + $state['delivery_count'] = ($state['delivery_count'] ?? 0) + 1; + $suffix = $state['delivery_count'] === 1 ? '' : '-'.$state['delivery_count']; + $state['identity'] = ['task_id' => 'task'.$suffix, 'workflow_task_id' => 'task'.$suffix, + 'activity_execution_id' => 'execution'.$suffix, 'activity_attempt_id' => 'attempt'.$suffix]; + $state['closed'] = false; + $state['expiry'] = $now + ($state['lease_seconds'] ?? 3); $state['progress_at'] = $now; $state['business_at'] = isset($state['business_seconds']) ? $now + $state['business_seconds'] : null; $deadlines = []; @@ -38,8 +44,7 @@ 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, + $response = ['task' => $state['identity'] + ['workflow_id' => 'workflow', 'run_id' => 'run', '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]], @@ -53,7 +58,7 @@ && ($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; + $state['expiry'] = $now + ($state['lease_seconds'] ?? 3); if ($heartbeat) { $state['progress_at'] = $now; $state['heartbeats'] = ($state['heartbeats'] ?? 0) + 1; @@ -61,8 +66,7 @@ $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', + $response = $state['identity'] + ['workflow_instance_id' => 'workflow', 'workflow_run_id' => 'run', '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), diff --git a/tests/WorkerActivityLeaseMaintenanceTest.php b/tests/WorkerActivityLeaseMaintenanceTest.php index 86545d1c..bbe1a12c 100644 --- a/tests/WorkerActivityLeaseMaintenanceTest.php +++ b/tests/WorkerActivityLeaseMaintenanceTest.php @@ -20,6 +20,8 @@ final class WorkerActivityLeaseMaintenanceTest extends TestCase private $tlsServer = null; private string $directory; private string $uri; + /** @var list */ + private array $workers = []; protected function setUp(): void { @@ -52,6 +54,10 @@ protected function setUp(): void protected function tearDown(): void { + foreach ($this->workers as $worker) { + $worker->requestShutdown(); + } + $this->workers = []; if (is_resource($this->tlsServer)) { proc_terminate($this->tlsServer); proc_close($this->tlsServer); @@ -71,6 +77,7 @@ protected function tearDown(): void public function testBlockingCallbackStaysInTheOriginalProcessAndOutlastsTheRuntimeLease(): void { $worker = $this->worker(); + (new \ReflectionProperty($worker, 'heartbeatIntervalSeconds'))->setValue($worker, 2); $captured = 0; $parent = getmypid(); $worker->registerActivity('slow', static function (ActivityContext $context) use (&$captured, $parent): int { @@ -100,6 +107,169 @@ public function testBlockingCallbackStaysInTheOriginalProcessAndOutlastsTheRunti self::assertSame($renewals, $this->state()['renewals'], 'Joined maintenance must stop after execution.'); } + public function testOneHelperServesDistinctActivitiesWithoutPerActivityBootstrapOrRenewal(): void + { + $worker = $this->worker(['activity_count' => 3, 'lease_seconds' => 6]); + $helpers = []; + $worker->registerActivity('slow', function (ActivityContext $context) use ($worker, &$helpers): int { + $helpers[] = $this->helperPid($worker); + return count($helpers); + }); + for ($activity = 0; $activity < 3; ++$activity) { + self::assertTrue($worker->tick(0)); + } + self::assertCount(1, array_unique($helpers)); + $state = $this->state(); + self::assertCount(3, $state['outcomes']); + self::assertSame(['attempt', 'attempt-2', 'attempt-3'], array_column($state['outcomes'], 'activity_attempt_id')); + self::assertSame(1, $state['renewals']); + self::assertCount(2, array_filter($state['requests'], + static fn (array $request): bool => $request['path'] === '/api/cluster/info')); + self::assertSame([], array_filter($state['requests'], + static fn (array $request): bool => $request['path'] === '/api/worker/heartbeat')); + } + + public function testAHelperKilledBetweenActivitiesIsReplacedAndProvesItsClientAgain(): void + { + if (!function_exists('posix_kill')) { + self::markTestSkipped('This helper-death control needs POSIX process observation.'); + } + $worker = $this->worker(['activity_count' => 2, 'lease_seconds' => 6]); + $worker->registerActivity('slow', static fn (ActivityContext $context): int => 1); + self::assertTrue($worker->tick(0)); + $helper = $this->helperPid($worker); + self::assertTrue(posix_kill($helper, 9)); + $handle = $this->helperHandle($worker); + $until = microtime(true) + 2; + while (proc_get_status($handle)['running'] && microtime(true) < $until) { + usleep(1000); + } + self::assertFalse(proc_get_status($handle)['running']); + self::assertTrue($worker->tick(0)); + self::assertNotSame($helper, $this->helperPid($worker)); + self::assertSame(2, $this->state()['renewals']); + self::assertCount(2, $this->state()['outcomes']); + } + + public function testWorkersWithTheSameClientDoNotShareAnIdleHelper(): void + { + $worker = $this->worker(['activity_count' => 2, 'lease_seconds' => 6]); + $other = new Worker(new Client($this->uri, namespace: 'tenant', workerToken: 'fixture-worker'), + 'queue', workerId: 'owner'); + $this->workers[] = $other; + foreach (['registered' => true, 'lastHeartbeatAt' => microtime(true), 'activityLeaseRenewalSupported' => true] as $name => $value) { + (new \ReflectionProperty($other, $name))->setValue($other, $value); + } + foreach ([$worker, $other] as $current) { + $current->registerActivity('slow', static fn (ActivityContext $context): int => 1); + self::assertTrue($current->tick(0)); + } + self::assertNotSame($this->helperPid($worker), $this->helperPid($other)); + } + + public function testShutdownClosesTheIdleHelper(): void + { + if (!function_exists('posix_kill')) { + self::markTestSkipped('This shutdown control needs POSIX process observation.'); + } + $worker = $this->worker(); + $worker->registerActivity('slow', static fn (ActivityContext $context): int => 1); + self::assertTrue($worker->tick(0)); + $helper = $this->helperPid($worker); + self::assertTrue(posix_kill($helper, 0)); + $worker->requestShutdown(); + self::assertFalse(posix_kill($helper, 0)); + } + + public function testForkedCallbackCannotUseOrTerminateItsParentsHelper(): void + { + if (!function_exists('pcntl_fork') || !function_exists('pcntl_waitpid')) { + self::markTestSkipped('This callback isolation control needs pcntl.'); + } + $worker = $this->worker(['activity_count' => 2, 'lease_seconds' => 6]); + $result = $this->directory.'/fork-result'; + $worker->registerActivity('slow', static function (ActivityContext $context) use ($worker, $result): int { + $supervisor = (new \ReflectionProperty($worker, 'activityLeaseProcess'))->getValue($worker); + $child = pcntl_fork(); + if ($child === 0) { + try { + $supervisor->check(); + file_put_contents($result, 'incorrectly accepted'); + } catch (\DurableWorkflow\Worker\WorkflowClaimRevoked $error) { + file_put_contents($result, $error->reason); + } finally { + $supervisor->stop(); + } + exit(0); + } + self::assertGreaterThan(0, $child); + pcntl_waitpid($child, $status); + self::assertTrue(pcntl_wifexited($status)); + self::assertSame(0, pcntl_wexitstatus($status)); + self::assertSame('activity_supervisor_process_mismatch', file_get_contents($result)); + return 1; + }); + self::assertTrue($worker->tick(0)); + $helper = $this->helperPid($worker); + self::assertTrue($worker->tick(0)); + self::assertSame($helper, $this->helperPid($worker)); + self::assertCount(2, $this->state()['outcomes']); + } + + public function testExpiredActivityAuthorityDoesNotCarryIntoTheNextClaim(): void + { + $worker = $this->worker(['activity_count' => 2, 'business_seconds' => 2]); + $calls = 0; + $worker->registerActivity('slow', static function (ActivityContext $context) use (&$calls): int { + if (++$calls === 1) { + usleep(2300000); + } + return $calls; + }); + self::assertTrue($worker->tick(0)); + self::assertSame([], $this->state()['outcomes'] ?? []); + self::assertTrue($worker->tick(0)); + self::assertSame(2, $calls); + self::assertCount(1, $this->state()['outcomes']); + self::assertSame('attempt-2', $this->state()['outcomes'][0]['activity_attempt_id']); + } + + public function testAnUnconfirmedFinishTerminatesTheHelper(): void + { + if (!function_exists('posix_kill') || !defined('SIGSTOP')) { + self::markTestSkipped('This lost-finish control needs POSIX stop signals.'); + } + $client = new Client($this->uri, 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, ['registered' => false, 'heartbeat_interval' => 10]); + $property = new \ReflectionProperty($process, 'process'); + $helper = proc_get_status($property->getValue($process))['pid']; + try { + self::assertTrue(posix_kill($helper, SIGSTOP)); + $started = hrtime(true); + $process->finish(); + self::assertLessThan(3, (hrtime(true) - $started) / 1e9); + self::assertNull($property->getValue($process)); + self::assertFalse(posix_kill($helper, 0)); + } finally { + $process->stop(); + } + } + + /** @return resource */ + private function helperHandle(Worker $worker) + { + $helper = (new \ReflectionProperty($worker, 'activityLeaseProcess'))->getValue($worker) + ?? (new \ReflectionProperty($worker, 'activityLeaseHelper'))->getValue($worker); + return (new \ReflectionProperty($helper, 'process'))->getValue($helper); + } + + private function helperPid(Worker $worker): int + { + return proc_get_status($this->helperHandle($worker))['pid']; + } + #[DataProvider('lateOutcomes')] public function testExpiredBusinessAuthorityFencesBothReturnAndFailure(bool $throw): void { @@ -363,6 +533,7 @@ private function worker(array $configuration = [], ?Client $client = null): Work (new \ReflectionProperty($worker, 'lastHeartbeatAt'))->setValue($worker, microtime(true)); (new \ReflectionMethod($worker, 'captureRegistrationFence'))->invoke($worker, ['server_capabilities' => ['activity_lease_renewal' => true]]); + $this->workers[] = $worker; return $worker; } diff --git a/tests/WorkerWorkflowTaskLeaseRetryTest.php b/tests/WorkerWorkflowTaskLeaseRetryTest.php index 382dc193..10fc448e 100644 --- a/tests/WorkerWorkflowTaskLeaseRetryTest.php +++ b/tests/WorkerWorkflowTaskLeaseRetryTest.php @@ -5,16 +5,254 @@ namespace DurableWorkflow\Tests; use DurableWorkflow\Client; +use DurableWorkflow\Codec\AvroPayloadCodec; use DurableWorkflow\Exception\ServerException; use DurableWorkflow\Exception\TransportException; use DurableWorkflow\Tests\Support\FakeTransport; +use DurableWorkflow\Transport\Psr18Transport; use DurableWorkflow\Worker; +use DurableWorkflow\Worker\ActivityContext; use DurableWorkflow\Worker\WorkflowContext; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; +use GuzzleHttp\Client as HttpClient; +use GuzzleHttp\Promise\FulfilledPromise; +use GuzzleHttp\Psr7\Response; +use Psr\Http\Message\RequestInterface; final class WorkerWorkflowTaskLeaseRetryTest extends TestCase { + #[DataProvider('workerWallClocks')] + public function testFreshWorkflowPollCompletesWithItsOriginalFenceWithoutInitialRenewal(float $wallClock): void + { + $transport = self::grantTransport(self::freshWorkflowPoll()); + $worker = self::grantWorker($transport, $wallClock); + + self::assertTrue($worker->tick(0)); + self::assertSame([], self::taskRequests($transport, 'heartbeat')); + $completions = self::taskRequests($transport, 'complete'); + self::assertCount(1, $completions); + self::assertSame('worker-1', $completions[0]['body']['lease_owner']); + self::assertSame(3, $completions[0]['body']['workflow_task_attempt']); + self::assertSame('complete_workflow', $completions[0]['body']['commands'][0]['type']); + } + + /** @return iterable */ + public static function workerWallClocks(): iterable + { + yield 'worker clock behind Server' => [0.0]; + yield 'worker clock ahead of Server' => [4_000_000_000.0]; + } + + /** @param array $poll */ + #[DataProvider('unqualifiedWorkflowPolls')] + public function testUnqualifiedWorkflowPollRetainsInitialRenewal(array $poll): void + { + $transport = self::grantTransport($poll); + + self::assertTrue(self::grantWorker($transport)->tick(0)); + self::assertCount(1, self::taskRequests($transport, 'heartbeat')); + self::assertCount(1, self::taskRequests($transport, 'complete')); + } + + /** @return iterable}> */ + public static function unqualifiedWorkflowPolls(): iterable + { + yield 'older Server' => [self::leasedWorkflowTask()]; + foreach ([null, 123, 'not-a-clock', '2026-07-18T00:00:00Z', + '2026-02-30T00:00:00.000000Z', "2026-07-18T00:00:00.000000Z\n"] as $index => $clock) { + yield 'invalid Server clock '.$index => [array_replace(self::freshWorkflowPoll(), ['server_time' => $clock])]; + } + foreach ([null, '2026-07-18T00:00:09.999999Z', '2026-07-18T00:00:00.000000Z', + '2026-07-17T23:59:59.000000Z', '2026-02-30T00:00:30.000000Z'] as $index => $expiry) { + $poll = self::freshWorkflowPoll(); + $poll['task']['lease_expires_at'] = $expiry; + yield 'unusable lease '.$index => [$poll]; + } + } + + public function testHistoryPagingTimeMakesThePollObservationStale(): void + { + $poll = self::freshWorkflowPoll(); + $poll['task']['next_history_page_token'] = 'history-page'; + $transport = self::grantTransport($poll, history: static function (): array { + usleep(1_100_000); + + return ['history_events' => [], 'next_history_page_token' => null]; + }); + + self::assertTrue(self::grantWorker($transport)->tick(0)); + self::assertCount(1, self::taskRequests($transport, 'heartbeat')); + self::assertCount(1, self::taskRequests($transport, 'complete')); + } + + public function testPollRoundTripConsumesTheRemainingServerLeaseMargin(): void + { + $poll = self::freshWorkflowPoll(); + $poll['task']['lease_expires_at'] = '2026-07-18T00:00:10.050000Z'; + $transport = self::grantTransport($poll, beforePoll: static function (): void { + usleep(120_000); + }); + + self::assertTrue(self::grantWorker($transport)->tick(0)); + self::assertCount(1, self::taskRequests($transport, 'heartbeat')); + } + + public function testFreshPollCompletionStillObeysTheServerFence(): void + { + $transport = self::grantTransport(self::freshWorkflowPoll(), completion: static function (): never { + throw self::leaseExpiredConflict(); + }); + + self::assertTrue(self::grantWorker($transport)->tick(0)); + self::assertSame([], self::taskRequests($transport, 'heartbeat')); + self::assertCount(1, self::taskRequests($transport, 'complete')); + self::assertSame([], self::taskRequests($transport, 'fail')); + } + + public function testFreshPollDoesNotSkipLocalActivityProgressRenewal(): void + { + $transport = self::grantTransport(self::freshWorkflowPoll()); + $worker = self::grantWorker($transport, handler: static fn (WorkflowContext $context): mixed => + $context->localActivity('local.receipt')); + $worker->registerActivity('local.receipt', static function (ActivityContext $context): string { + $context->heartbeat(['step' => 1]); + + return 'done'; + }); + self::assertTrue($worker->tick(0)); + $heartbeats = self::taskRequests($transport, 'heartbeat'); + self::assertCount(1, $heartbeats); + self::assertSame(['lease_owner' => 'worker-1', 'workflow_task_attempt' => 3], $heartbeats[0]['body']); + $completion = self::taskRequests($transport, 'complete'); + self::assertSame('record_local_activity', $completion[0]['body']['commands'][0]['type']); + } + + public function testShutdownDuringHistoryLoadingCannotUseTheFreshGrant(): void + { + $poll = self::freshWorkflowPoll(); + $poll['task']['next_history_page_token'] = 'history-page'; + $worker = null; + $executions = 0; + $transport = self::grantTransport($poll, history: static function () use (&$worker): array { + $worker->requestShutdown(); + + return ['history_events' => [], 'next_history_page_token' => null]; + }); + $worker = self::grantWorker($transport, handler: static function (WorkflowContext $context) use (&$executions): string { + ++$executions; + + return 'must-not-run'; + }); + + self::assertTrue($worker->tick(0)); + self::assertSame(0, $executions); + self::assertSame([], self::taskRequests($transport, 'heartbeat')); + self::assertSame([], self::taskRequests($transport, 'complete')); + } + + public function testExternalPayloadLoadingCannotRefreshThePollObservation(): void + { + $blob = (new AvroPayloadCodec())->envelope(['loaded'])['blob']; + $events = []; + $http = new HttpClient(['handler' => static function (RequestInterface $request) use ($blob, &$events): FulfilledPromise { + $path = $request->getUri()->getPath(); + if (str_contains($path, '/external-payloads/')) { + $events[] = 'fetch'; + usleep(1_100_000); + + return new FulfilledPromise(new Response(200, [ + 'Content-Type' => 'application/octet-stream', + 'X-Durable-Workflow-Payload-Codec' => 'avro', + 'X-Durable-Workflow-Payload-Size' => (string) strlen($blob), + 'X-Durable-Workflow-Payload-SHA256' => hash('sha256', $blob), + ], $blob)); + } + if (str_ends_with($path, '/workflow-tasks/poll')) { + $reply = self::freshWorkflowPoll(); + $reply['task']['arguments'] = ['codec' => 'avro', 'external_payload' => RuntimeExternalPayloadTest::reference($blob)]; + } elseif (str_ends_with($path, '/heartbeat')) { + $events[] = 'heartbeat'; + $reply = self::renewedLease(); + } elseif (str_ends_with($path, '/complete')) { + $events[] = 'complete'; + $reply = ['completed' => true]; + } else { + $reply = ['task' => null, 'poll_status' => 'empty']; + } + + return new FulfilledPromise(new Response(200, [], json_encode($reply, JSON_THROW_ON_ERROR))); + }]); + $worker = new Worker(new Client('https://server.example', transport: new Psr18Transport($http)), + 'orders', workerId: 'worker-1'); + $worker->registerWorkflow('orders.workflow', static function (WorkflowContext $context, string $input) use (&$events): string { + self::assertSame('loaded', $input); + $events[] = 'handler'; + + return 'completed'; + }); + + self::assertTrue($worker->tick(0)); + self::assertSame(['heartbeat', 'handler', 'complete'], array_slice($events, -3)); + } + + /** @return array */ + private static function freshWorkflowPoll(): array + { + $poll = self::leasedWorkflowTask(); + $poll['server_time'] = '2026-07-18T00:00:00.000000Z'; + $poll['task']['lease_expires_at'] = '2026-07-18T00:00:30.000000Z'; + + return $poll; + } + + /** @param array $poll */ + private static function grantTransport( + array $poll, + ?\Closure $beforePoll = null, + ?\Closure $history = null, + ?\Closure $completion = null, + ): FakeTransport { + return new FakeTransport(handler: static function (string $method, string $uri) use ( + $poll, $beforePoll, $history, $completion, + ): array { + if (str_ends_with($uri, '/workflow-tasks/poll')) { + $beforePoll?->__invoke(); + + return $poll; + } + if (str_ends_with($uri, '/workflow-task-1/heartbeat')) { + return self::renewedLease(); + } + if (str_ends_with($uri, '/workflow-task-1/history') && $history !== null) { + return $history(); + } + if (str_ends_with($uri, '/workflow-task-1/complete')) { + return $completion === null ? ['completed' => true] : $completion(); + } + if (str_ends_with($uri, '/poll')) { + return ['task' => null, 'poll_status' => 'empty']; + } + + self::fail("Unexpected worker request: {$method} {$uri}"); + }); + } + + private static function grantWorker(FakeTransport $transport, float $wallClock = 0.0, ?\Closure $handler = null): Worker + { + $worker = new Worker(new Client('https://server.example', transport: $transport), 'orders', + workerId: 'worker-1', clock: static fn (): float => $wallClock); + + return $worker->registerWorkflow('orders.workflow', $handler ?? static fn (WorkflowContext $context): string => 'completed'); + } + + /** @return list, body: ?array}> */ + private static function taskRequests(FakeTransport $transport, string $operation): array + { + return array_values(array_filter($transport->requests, + static fn (array $request): bool => str_ends_with($request['uri'], '/workflow-task-1/'.$operation))); + } + public function testTransientLeaseRenewalThenSuccessExecutesAndCompletesExactlyOnce(): void { $now = 0.0; diff --git a/tests/WorkflowPollGrantTest.php b/tests/WorkflowPollGrantTest.php new file mode 100644 index 00000000..b336a833 --- /dev/null +++ b/tests/WorkflowPollGrantTest.php @@ -0,0 +1,75 @@ +isFreshFor($poll['task'], 0.2)); + $poll['task'][$field] = $changed; + self::assertFalse($grant->isFreshFor($poll['task'], 0.2)); + } + + /** @return iterable */ + public static function grantFenceChanges(): iterable + { + yield 'other task' => ['task_id', 'workflow-task-2']; + yield 'other owner' => ['lease_owner', 'worker-2']; + yield 'other attempt' => ['workflow_task_attempt', 4]; + yield 'attempt type alias' => ['workflow_task_attempt', '3']; + yield 'changed expiry' => ['lease_expires_at', '2026-07-18T00:01:00.000000Z']; + } + + #[DataProvider('monotonicGrantBounds')] + public function testServerLeaseAndMonotonicObservationBothBoundFreshness(float $received, float $now, bool $fresh): void + { + $poll = self::freshWorkflowPoll(); + $grant = WorkflowPollGrant::observe($poll, 'worker-1', 0, $received); + self::assertNotNull($grant); + self::assertSame($fresh, $grant->isFreshFor($poll['task'], $now)); + } + + /** @return iterable */ + public static function monotonicGrantBounds(): iterable + { + yield 'fresh with margin' => [0.1, 0.2, true]; + yield 'exactly one second old' => [0.0, 1.0, false]; + yield 'backwards monotonic observation' => [0.2, 0.1, false]; + yield 'exactly ten seconds remain including request' => [19.5, 20.0, true]; + yield 'request time leaves under ten seconds' => [20.0, 20.1, false]; + yield 'nonfinite current reading' => [0.0, INF, false]; + } + + public function testInvalidOrForeignPollIdentityCannotCreateFreshAuthority(): void + { + $poll = self::freshWorkflowPoll(); + foreach (['task_id' => '', 'lease_owner' => 'other-worker', 'workflow_task_attempt' => '3'] as $key => $value) { + $invalid = $poll; + $invalid['task'][$key] = $value; + self::assertNull(WorkflowPollGrant::observe($invalid, 'worker-1', 0, 0.1)); + } + self::assertNull(WorkflowPollGrant::observe(['task' => null], 'worker-1', 0, 0.1)); + self::assertNull(WorkflowPollGrant::observe($poll, 'worker-1', 0.2, 0.1)); + self::assertNull(WorkflowPollGrant::observe($poll, 'worker-1', NAN, 0.1)); + self::assertNull(WorkflowPollGrant::observe($poll, 'worker-1', 0, INF)); + } + + /** @return array */ + private static function freshWorkflowPoll(): array + { + return ['poll_status' => 'leased', 'server_time' => '2026-07-18T00:00:00.000000Z', + 'task' => ['task_id' => 'workflow-task-1', 'lease_owner' => 'worker-1', 'workflow_task_attempt' => 3, + 'lease_expires_at' => '2026-07-18T00:00:30.000000Z']]; + } +} diff --git a/tests/fixtures/replay-regressions/workflow-poll-fresh-grant.json b/tests/fixtures/replay-regressions/workflow-poll-fresh-grant.json new file mode 100644 index 00000000..f8051a94 --- /dev/null +++ b/tests/fixtures/replay-regressions/workflow-poll-fresh-grant.json @@ -0,0 +1,24 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "php-workflow-poll-fresh-grant", + "protocol_version": "1.20", + "bindings": [ + "php" + ], + "workflow": { + "type": "golden.workflow-fresh-grant", + "input": [] + }, + "expected": { + "type": "complete_workflow", + "result": "fresh-grant", + "initial_workflow_renewals": 0 + }, + "command_sequence": [ + { + "type": "complete_workflow", + "result": "fresh-grant" + } + ] +}