Backport activity grant transit correction to 3.9.x

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)) {

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);

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<string, mixed> $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<string, mixed> $task
      * @return array<string, string|int>
      */

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;
         }

src/Worker/ActivityLeaseProcess.php

@@ -34,7 +34,7 @@ public static function available(): bool
     private array $identity = [];
 
     /** @param array<string, mixed> $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,15 +44,16 @@ public function __construct(Client $client, array $task, string $owner, string $
     }
 
     /** @param array<string, mixed> $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
     {
         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();

tests/Support/ActivityGrantScenario.php

@@ -0,0 +1,124 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Tests\Support;
+
+use Composer\Autoload\ClassLoader;
+use DurableWorkflow\Client;
+use DurableWorkflow\Codec\AvroPayloadCodec;
+use DurableWorkflow\Exception\ServerException;
+use DurableWorkflow\Worker;
+use DurableWorkflow\Worker\ActivityContext;
+use ReflectionClass;
+use ReflectionMethod;
+use ReflectionProperty;
+use RuntimeException;
+
+/** An inert HTTP delivery delay exercised through the public Worker. */
+final class ActivityGrantScenario
+{
+    /** @return array{callbacks: int, commands: list<array<string, mixed>>} */
+    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<array<string, mixed>>}
+     */
+    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 = "<?php\nrequire ".var_export($vendor.'/autoload.php', true).";\n"
+            .'$loader = new \Composer\Autoload\ClassLoader(__DIR__);'."\n"
+            .'$source = '.var_export($source.'/src', true).";\n"
+            .'$loader->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);
+        }
+    }
+}

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 === []) {

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);

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')) {

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
+    }
+  ]
+}