From 78f25360049071bafe1dc8e521a75b48fff4ec29 Mon Sep 17 00:00:00 2001 From: Richard McDaniel Date: Fri, 6 Nov 2026 15:36:00 -0600 Subject: [PATCH] Backport activity poll transit correction to 3.9.x (cherry picked from commit c87eae373325d0a5bc8529058984becf17c82793) --- scripts/ci/run-replay-regression-fixture.php | 10 ++ src/Worker.php | 8 +- src/Worker/ActivityLeaseAuthority.php | 26 ++++ src/Worker/ActivityLeaseMaintainer.php | 3 +- src/Worker/ActivityLeaseProcess.php | 10 +- tests/Support/ActivityGrantScenario.php | 124 ++++++++++++++++++ tests/Support/ReplayRegressionFixture.php | 9 ++ tests/Support/activity-lease-http.php | 4 + tests/WorkerActivityLeaseMaintenanceTest.php | 19 +++ .../activity-poll-transit-consumes-grant.json | 23 ++++ 10 files changed, 230 insertions(+), 6 deletions(-) create mode 100644 tests/Support/ActivityGrantScenario.php create mode 100644 tests/fixtures/replay-regressions/activity-poll-transit-consumes-grant.json diff --git a/scripts/ci/run-replay-regression-fixture.php b/scripts/ci/run-replay-regression-fixture.php index 59fa26f5..427bc74a 100644 --- a/scripts/ci/run-replay-regression-fixture.php +++ b/scripts/ci/run-replay-regression-fixture.php @@ -442,6 +442,16 @@ private static function execute(array $fixture): void throw new RuntimeException("{$identity}.history must not be empty."); } + if ($workflowType === 'golden.activity-poll-transit') { + require_once __DIR__.'/../../tests/Support/ActivityGrantScenario.php'; + $observed = \DurableWorkflow\Tests\Support\ActivityGrantScenario::runIsolated(); + self::assertMatches($fixture['command_sequence'], $observed['commands'], $identity.'.command_sequence'); + self::assertMatches($fixture['expected'], $observed['commands'][0] + + ['activity_callback_count' => $observed['callbacks']], $identity.'.expected'); + + return; + } + $workflowTasks = $fixture['workflow_tasks'] ?? null; if ($workflowTasks !== null) { if (array_key_exists('command_sequence', $fixture)) { diff --git a/src/Worker.php b/src/Worker.php index 33ef5199..ab0db29e 100644 --- a/src/Worker.php +++ b/src/Worker.php @@ -116,6 +116,7 @@ final class Worker private ?ActivityLeaseProcess $activityLeaseHelper = null; private ?WorkflowPollGrant $workflowPollGrant = null; private ?float $workflowPollStartedAt = null; + private ?float $activityPollStartedAt = null; /** @var (\Closure(): void)|null */ private ?\Closure $unbindActivityLease = null; private ?CancellationRequest $claimCancellation = null; @@ -1150,6 +1151,9 @@ private function transientRetryDelay(int $attempt, mixed $retryAfterSeconds): fl */ private function observePoll(string $taskKind, \Closure $operation): array { + if ($taskKind === 'activity') { + $this->activityPollStartedAt = hrtime(true) / 1e9; + } if ($taskKind === 'workflow') { $this->workflowPollGrant = null; $this->workflowPollStartedAt = hrtime(true) / 1e9; @@ -1623,7 +1627,9 @@ private function workerPayloadProgress(string $path, ?array $body, array $respon throw new WorkflowClaimRevoked('activity_supervisor_conflict', 'A previous activity supervisor is still active.'); } $presence = ['registered' => $this->registered, 'heartbeat_interval' => $this->heartbeatIntervalSeconds, - 'presence_due_in' => max(0.0, $this->heartbeatIntervalSeconds - $this->elapsedSinceHeartbeat())]; + 'presence_due_in' => max(0.0, $this->heartbeatIntervalSeconds - $this->elapsedSinceHeartbeat()), + 'poll_started_at' => $this->activityPollStartedAt]; + $this->activityPollStartedAt = null; if ($this->activityLeaseHelper === null) { $this->activityLeaseHelper = new ActivityLeaseProcess($this->client, $task, $this->workerId, $this->taskQueue, $response['server_time'] ?? null, hrtime(true) / 1e9, $presence); diff --git a/src/Worker/ActivityLeaseAuthority.php b/src/Worker/ActivityLeaseAuthority.php index f4ff093f..5bce1789 100644 --- a/src/Worker/ActivityLeaseAuthority.php +++ b/src/Worker/ActivityLeaseAuthority.php @@ -57,6 +57,32 @@ public function __construct( $this->remaining(); } + /** Unknown or slow deliveries retain confirmation; a fast grant charges the + * entire poll before its lease, business and heartbeat bounds are captured. + * @param array $task + * @return array{self, float, bool} + */ + public static function capturePollGrant(array $task, string $namespace, string $owner, string $queue, + mixed $serverTime, float $received, mixed $started): array + { + $authority = new self($task, $namespace, $owner, $queue, $serverTime, $received); + if ((!is_int($started) && !is_float($started)) || !is_finite($started) + || $started < 0 || $started > $received || $received - $started >= 1) { + return [$authority, $received, false]; + } + try { + $original = new self($task, $namespace, $owner, $queue, $serverTime, (float) $started); + if ($original->remaining() >= 3) { + return [$original, (float) $started, true]; + } + } catch (WorkflowClaimRevoked) { + // The response can arrive after the conservative original bound. + // A renewal must confirm it before an application callback starts. + } + + return [$authority, $received, false]; + } + /** @param array $task * @return array */ diff --git a/src/Worker/ActivityLeaseMaintainer.php b/src/Worker/ActivityLeaseMaintainer.php index b1a0b191..dadb7e2f 100644 --- a/src/Worker/ActivityLeaseMaintainer.php +++ b/src/Worker/ActivityLeaseMaintainer.php @@ -243,7 +243,8 @@ private function applyRenewal(?array $reply, float $started): 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) { + if ($proven && ($this->start['fresh_poll_grant'] ?? false) === true + && $age >= 0 && $age < 1 && $this->authority->remaining() >= 3) { $this->scheduleRenewal(); return; } diff --git a/src/Worker/ActivityLeaseProcess.php b/src/Worker/ActivityLeaseProcess.php index cd3d4906..ad50ce39 100644 --- a/src/Worker/ActivityLeaseProcess.php +++ b/src/Worker/ActivityLeaseProcess.php @@ -34,7 +34,7 @@ public static function available(): bool private array $identity = []; /** @param array $task - * @param array{registered: bool, heartbeat_interval: int, presence_due_in?: float} $presence + * @param array{registered: bool, heartbeat_interval: int, presence_due_in?: float, poll_started_at?: float|null} $presence */ public function __construct(Client $client, array $task, string $owner, string $queue, mixed $serverTime, float $captured, array $presence) @@ -44,7 +44,7 @@ public function __construct(Client $client, array $task, string $owner, string $ } /** @param array $task - * @param array{registered: bool, heartbeat_interval: int, presence_due_in?: float} $presence + * @param array{registered: bool, heartbeat_interval: int, presence_due_in?: float, poll_started_at?: float|null} $presence */ public function start(Client $client, array $task, string $owner, string $queue, mixed $serverTime, float $captured, array $presence): void @@ -52,7 +52,8 @@ public function start(Client $client, array $task, string $owner, string $queue, 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); + [$authority, $captured, $fresh] = ActivityLeaseAuthority::capturePollGrant($task, $client->namespace, $owner, $queue, + $serverTime, $captured, $presence['poll_started_at'] ?? null); $this->identity = $authority->identity(); $this->until = hrtime(true) / 1e9 + $authority->remaining(); $this->fixedUntil = $authority->fixedBound(); @@ -92,7 +93,8 @@ public function start(Client $client, array $task, string $owner, string $queue, 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, + 'server_time' => $serverTime, 'captured_at' => $captured, 'fresh_poll_grant' => $fresh, + 'owner' => $owner, 'queue' => $queue, 'parent_pid' => getmypid()] + $presence); } catch (Throwable $error) { $this->stop(); diff --git a/tests/Support/ActivityGrantScenario.php b/tests/Support/ActivityGrantScenario.php new file mode 100644 index 00000000..0d0ea6ba --- /dev/null +++ b/tests/Support/ActivityGrantScenario.php @@ -0,0 +1,124 @@ +>} */ + public static function run(string $uri, string $statePath): array + { + file_put_contents($statePath, json_encode(['activity_count' => 2, 'lease_seconds' => 4, + 'delay_poll_delivery' => 2, 'poll_reply_delay_micros' => 4_250_000], JSON_THROW_ON_ERROR)); + $worker = new Worker(new Client($uri, namespace: 'tenant', workerToken: 'fixture-worker'), + 'queue', workerId: 'owner'); + (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]]); + $calls = 0; + $worker->registerActivity('slow', static function (ActivityContext $context) use (&$calls): int { return ++$calls; }); + try { + if (!$worker->tick(0)) { + throw new RuntimeException('The original activity fixture task was not executed.'); + } + try { + $worker->tick(0); + } catch (ServerException $error) { + if ($error->status !== 409) { + throw $error; + } + } + $state = json_decode((string) file_get_contents($statePath), true, flags: JSON_THROW_ON_ERROR); + $codec = new AvroPayloadCodec(); + $commands = array_map(static fn (array $body): array => ['type' => 'complete_activity', + 'activity_attempt_id' => $body['activity_attempt_id'], 'result' => $codec->decodeEnvelope($body['result'])], + $state['outcomes'] ?? []); + + return ['callbacks' => $calls, 'commands' => $commands]; + } finally { + $worker->requestShutdown(); + } + } + + /** Parent and helper must load the selected immutable SDK, not a cached vendor's SDK. + * @return array{callbacks: int, commands: list>} + */ + public static function runIsolated(): array + { + if (!function_exists('proc_open') || !extension_loaded('curl')) { + throw new RuntimeException('The activity grant fixture requires CLI proc_open and cURL.'); + } + $directory = sys_get_temp_dir().'/activity-grant-'.bin2hex(random_bytes(8)); + mkdir($directory, 0700); + mkdir($directory.'/vendor', 0700); + $vendor = dirname((new ReflectionClass(ClassLoader::class))->getFileName(), 2); + $source = dirname((new ReflectionClass(Worker::class))->getFileName(), 2); + $bootstrap = "setPsr4("DurableWorkflow\\\\", $source); $loader->register(true);'."\n" + .'foreach (["Client", "Worker", "Worker/ActivityLeaseProcess", "Worker/ActivityLeaseMaintainer"] as $name) {'."\n" + .' $file = (new \ReflectionClass("DurableWorkflow\\\\".str_replace("/", "\\\\", $name)))->getFileName();'."\n" + .' if (realpath($file) !== realpath($source."/".$name.".php")) {'."\n" + .' throw new \RuntimeException("Activity fixture loaded a different SDK source.");'."\n" + ." }\n}\nreturn \$loader;\n"; + file_put_contents($directory.'/vendor/autoload.php', $bootstrap); + $loader = null; + $server = null; + try { + $loader = require $directory.'/vendor/autoload.php'; + $socket = stream_socket_server('tcp://127.0.0.1:0'); + if (!is_resource($socket)) { + throw new RuntimeException('The activity fixture cannot reserve a loopback address.'); + } + $address = stream_socket_get_name($socket, false); + fclose($socket); + $server = proc_open([PHP_BINARY, '-S', $address, __DIR__.'/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' => $directory.'/state.json']); + if (!is_resource($server)) { + throw new RuntimeException('The activity fixture HTTP process did not start.'); + } + $until = microtime(true) + 5; + do { + $ready = @stream_socket_client('tcp://'.$address, $errno, $message, 0.1); + if (is_resource($ready)) { + fclose($ready); + + return self::run('http://'.$address, $directory.'/state.json'); + } + usleep(10000); + } while (microtime(true) < $until); + throw new RuntimeException('The activity fixture HTTP process was not ready.'); + } finally { + if (is_resource($server)) { + proc_terminate($server); + proc_close($server); + } + if ($loader instanceof ClassLoader) { + $loader->unregister(); + } + foreach ([$directory.'/state.json', $directory.'/vendor/autoload.php'] as $path) { + if (is_file($path)) { + unlink($path); + } + } + rmdir($directory.'/vendor'); + rmdir($directory); + } + } +} diff --git a/tests/Support/ReplayRegressionFixture.php b/tests/Support/ReplayRegressionFixture.php index 2c8b21e2..475bd3f6 100644 --- a/tests/Support/ReplayRegressionFixture.php +++ b/tests/Support/ReplayRegressionFixture.php @@ -86,6 +86,15 @@ public static function execute(array $fixture): array return self::executeFreshWorkflowPoll($identity, $input, $history, $fixture); } + if ($workflowType === 'golden.activity-poll-transit') { + $observed = ActivityGrantScenario::runIsolated(); + self::assertMatches($fixture['command_sequence'], $observed['commands'], $identity.'.command_sequence'); + self::assertMatches($fixture['expected'], $observed['commands'][0] + + ['activity_callback_count' => $observed['callbacks']], $identity.'.expected'); + + return $observed['commands']; + } + $workflowTasks = $fixture['workflow_tasks'] ?? null; if ($workflowTasks !== null) { if (!is_array($workflowTasks) || !array_is_list($workflowTasks) || $workflowTasks === []) { diff --git a/tests/Support/activity-lease-http.php b/tests/Support/activity-lease-http.php index c0d14556..c1a65818 100644 --- a/tests/Support/activity-lease-http.php +++ b/tests/Support/activity-lease-http.php @@ -114,4 +114,8 @@ fclose($handle); http_response_code($status); header('Content-Type: application/json'); +if (($response['poll_status'] ?? null) === 'leased' + && ($state['delivery_count'] ?? 0) === ($state['delay_poll_delivery'] ?? null)) { + usleep($state['poll_reply_delay_micros'] ?? 0); +} echo json_encode($response, JSON_THROW_ON_ERROR); diff --git a/tests/WorkerActivityLeaseMaintenanceTest.php b/tests/WorkerActivityLeaseMaintenanceTest.php index bbe1a12c..fae9dcb6 100644 --- a/tests/WorkerActivityLeaseMaintenanceTest.php +++ b/tests/WorkerActivityLeaseMaintenanceTest.php @@ -5,6 +5,7 @@ namespace DurableWorkflow\Tests; use DurableWorkflow\Client; +use DurableWorkflow\Tests\Support\ActivityGrantScenario; use DurableWorkflow\Worker; use DurableWorkflow\Worker\ActivityContext; use DurableWorkflow\Transport\Psr18Transport; @@ -129,6 +130,24 @@ public function testOneHelperServesDistinctActivitiesWithoutPerActivityBootstrap static fn (array $request): bool => $request['path'] === '/api/worker/heartbeat')); } + public function testTransportDelayCannotMakeAnExpiredGrantLookFreshToAReusedHelper(): void + { + $observed = ActivityGrantScenario::run($this->uri, $this->directory.'/state.json'); + self::assertSame(1, $observed['callbacks'], 'A grant consumed by response transit cannot authorize another callback.'); + self::assertCount(1, $observed['commands']); + } + + public function testPollTransitConsumesTheFreshGrantMargin(): void + { + $worker = $this->worker(['activity_count' => 2, 'lease_seconds' => 3.4, + 'delay_poll_delivery' => 2, 'poll_reply_delay_micros' => 600_000]); + $worker->registerActivity('slow', static fn (ActivityContext $context): int => 1); + self::assertTrue($worker->tick(0)); + self::assertTrue($worker->tick(0)); + self::assertCount(2, $this->state()['outcomes']); + self::assertSame(2, $this->state()['renewals'], 'A short remaining grant needs confirmation after charging transit.'); + } + public function testAHelperKilledBetweenActivitiesIsReplacedAndProvesItsClientAgain(): void { if (!function_exists('posix_kill')) { diff --git a/tests/fixtures/replay-regressions/activity-poll-transit-consumes-grant.json b/tests/fixtures/replay-regressions/activity-poll-transit-consumes-grant.json new file mode 100644 index 00000000..32f57c72 --- /dev/null +++ b/tests/fixtures/replay-regressions/activity-poll-transit-consumes-grant.json @@ -0,0 +1,23 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "php-activity-poll-transit-consumes-grant", + "protocol_version": "1.20", + "bindings": ["php"], + "workflow": { + "type": "golden.activity-poll-transit", + "input": [] + }, + "expected": { + "type": "complete_activity", + "result": 1, + "activity_callback_count": 1 + }, + "command_sequence": [ + { + "type": "complete_activity", + "activity_attempt_id": "attempt", + "result": 1 + } + ] +}