From 09ee765798559a2082a71cdcda3d07dfde10cde3 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Sun, 5 Jul 2026 10:11:33 +0000 Subject: [PATCH 01/37] Implement transparent routing to Cloud Tasks for Push Queues --- composer.json | 3 +- src/Api/TaskQueue/PushQueue.php | 137 ++++++++++++++++++++++++++++++++ src/Api/TaskQueue/PushTask.php | 10 +++ 3 files changed, 149 insertions(+), 1 deletion(-) diff --git a/composer.json b/composer.json index 87438a71..bb3d7ae3 100644 --- a/composer.json +++ b/composer.json @@ -13,7 +13,8 @@ "php": ">=7.2.0", "guzzlehttp/streams": "^3.0", "guzzlehttp/guzzle": "^7.2", - "composer/semver": "^3.2" + "composer/semver": "^3.2", + "google/cloud-tasks": "^1.0" }, "require-dev": { "phpunit/phpunit": "^8", diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index fdfa520f..913975a5 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -147,6 +147,11 @@ public function addTasks($tasks) { '$tasks must contain at most ' . self::MAX_TASKS_PER_ADD . ' tasks. Actual size: ' . count($tasks)); } + + if (getenv('GAE_PUSHQUEUE_BACKEND') === 'CLOUD_TASK') { + return $this->addTasksCloudTasks($tasks); + } + $req = new TaskQueueBulkAddRequest(); $resp = new TaskQueueBulkAddResponse(); @@ -212,4 +217,136 @@ public function addTasks($tasks) { } return $names; } + + private static function getMetadataValue($path) { + $opts = [ + 'http' => [ + 'method' => 'GET', + 'header' => 'Metadata-Flavor: Google', + 'timeout' => 1.0 + ] + ]; + $context = stream_context_create($opts); + $url = 'http://metadata.google.internal/computeMetadata/v1/' . $path; + $result = @file_get_contents($url, false, $context); + return $result; + } + + private static function getRegion() { + static $region = null; + if ($region === null) { + $region = getenv('REGION_ID'); + if (!$region) { + $zone = self::getMetadataValue('instance/zone'); + if ($zone) { + $parts = explode('/', $zone); + $zoneName = end($parts); + $dashPos = strrpos($zoneName, '-'); + if ($dashPos !== false) { + $region = substr($zoneName, 0, $dashPos); + } else { + $region = $zoneName; + } + } + } + if (!$region) { + $region = 'us-central1'; + } + } + return $region; + } + + private static function getProjectId() { + static $projectId = null; + if ($projectId === null) { + $projectId = getenv('GOOGLE_CLOUD_PROJECT'); + if (!$projectId) { + $projectId = self::getMetadataValue('project/project-id'); + } + if (!$projectId) { + $appId = ApiProxy::getCurrentAppId(); + if (($pos = strpos($appId, '~')) !== false) { + $projectId = substr($appId, $pos + 1); + } else { + $projectId = $appId; + } + } + } + return $projectId; + } + + private function addTasksCloudTasks($tasks) { + $client = new \Google\Cloud\Tasks\V2\CloudTasksClient(); + $projectId = self::getProjectId(); + $region = self::getRegion(); + $queueName = $client->queueName($projectId, $region, $this->name); + + $names = []; + foreach ($tasks as $task) { + $ctTask = new \Google\Cloud\Tasks\V2\Task(); + + if ($task->getName()) { + $ctTask->setName($client->taskName($projectId, $region, $this->name, $task->getName())); + } + + $httpRequest = new \Google\Cloud\Tasks\V2\HttpRequest(); + + $url = $task->getUrl(); + if (strncmp($url, '/', 1) === 0) { + $hostname = \Google\AppEngine\Api\Modules\ModulesService::getHostname(); + $url = "https://" . $hostname . $url; + } + $httpRequest->setUrl($url); + + $methodStr = $task->getMethod(); + $methodMap = [ + 'POST' => \Google\Cloud\Tasks\V2\HttpMethod::POST, + 'GET' => \Google\Cloud\Tasks\V2\HttpMethod::GET, + 'HEAD' => \Google\Cloud\Tasks\V2\HttpMethod::HEAD, + 'PUT' => \Google\Cloud\Tasks\V2\HttpMethod::PUT, + 'DELETE' => \Google\Cloud\Tasks\V2\HttpMethod::DELETE, + ]; + $httpRequest->setHttpMethod($methodMap[$methodStr]); + + $headers = []; + foreach ($task->getHeaders() as $header) { + $pair = explode(':', $header, 2); + $headers[trim($pair[0])] = trim($pair[1]); + } + $httpRequest->setHeaders($headers); + + if ($methodStr === 'POST' || $methodStr === 'PUT') { + if ($task->getQueryData()) { + $httpRequest->setBody(http_build_query($task->getQueryData())); + } + } + + $ctTask->setHttpRequest($httpRequest); + + if ($task->getDelaySeconds() > 0) { + $scheduleTime = new \Google\Protobuf\Timestamp(); + $scheduleTime->setSeconds(time() + $task->getDelaySeconds()); + $ctTask->setScheduleTime($scheduleTime); + } + + $request = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) + ->setParent($queueName) + ->setTask($ctTask); + + try { + $response = $client->createTask($request); + $fullName = $response->getName(); + $parts = explode('/', $fullName); + $names[] = end($parts); + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS') { + throw new TaskAlreadyExistsException('Task with the same name exists already'); + } + throw new TaskQueueException('Cloud Tasks Error: ' . $e->getMessage(), $e->getCode()); + } catch (\Exception $e) { + throw new TaskQueueException('Error calling Cloud Tasks: ' . $e->getMessage()); + } + } + return $names; + } } diff --git a/src/Api/TaskQueue/PushTask.php b/src/Api/TaskQueue/PushTask.php index e0409810..1c39a74e 100644 --- a/src/Api/TaskQueue/PushTask.php +++ b/src/Api/TaskQueue/PushTask.php @@ -64,6 +64,7 @@ final class PushTask { 'method' => 'POST', 'name' => '', 'header' => '', + 'transactional' => false, ]; private $url; @@ -287,6 +288,15 @@ public function getHeaders() { * exists in the queue. * @throws TaskQueueException if there was a problem using the service. */ + /** + * Return whether the task is transactional. + * + * @return bool Whether the task is transactional. + */ + public function isTransactional() { + return $this->options['transactional']; + } + public function add($queue_name = 'default') { $queue = new PushQueue($queue_name); return $queue->addTasks([$this])[0]; From 86c0e089822210680c1e6222d9f214fe2a8fa575 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Sun, 5 Jul 2026 10:13:02 +0000 Subject: [PATCH 02/37] Remove transactional task support from PHP SDK as it is out of scope --- src/Api/TaskQueue/PushTask.php | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/src/Api/TaskQueue/PushTask.php b/src/Api/TaskQueue/PushTask.php index 1c39a74e..96d03248 100644 --- a/src/Api/TaskQueue/PushTask.php +++ b/src/Api/TaskQueue/PushTask.php @@ -64,7 +64,6 @@ final class PushTask { 'method' => 'POST', 'name' => '', 'header' => '', - 'transactional' => false, ]; private $url; @@ -288,16 +287,7 @@ public function getHeaders() { * exists in the queue. * @throws TaskQueueException if there was a problem using the service. */ - /** - * Return whether the task is transactional. - * - * @return bool Whether the task is transactional. - */ - public function isTransactional() { - return $this->options['transactional']; - } - - public function add($queue_name = 'default') { + public function add($queue_name = 'default') { $queue = new PushQueue($queue_name); return $queue->addTasks([$this])[0]; } From 6afb1d5bb18ffd09f350dd5536a5aee9d62b49cf Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Sun, 5 Jul 2026 10:13:58 +0000 Subject: [PATCH 03/37] Support Host header routing in Cloud Tasks URL resolution --- src/Api/TaskQueue/PushQueue.php | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 913975a5..f997a016 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -291,9 +291,22 @@ private function addTasksCloudTasks($tasks) { $httpRequest = new \Google\Cloud\Tasks\V2\HttpRequest(); + $headers = []; + $hostHeader = null; + foreach ($task->getHeaders() as $header) { + $pair = explode(':', $header, 2); + $key = trim($pair[0]); + $val = trim($pair[1]); + $headers[$key] = $val; + if (strcasecmp($key, 'Host') === 0) { + $hostHeader = $val; + } + } + $httpRequest->setHeaders($headers); + $url = $task->getUrl(); if (strncmp($url, '/', 1) === 0) { - $hostname = \Google\AppEngine\Api\Modules\ModulesService::getHostname(); + $hostname = $hostHeader ?: \Google\AppEngine\Api\Modules\ModulesService::getHostname(); $url = "https://" . $hostname . $url; } $httpRequest->setUrl($url); @@ -308,13 +321,6 @@ private function addTasksCloudTasks($tasks) { ]; $httpRequest->setHttpMethod($methodMap[$methodStr]); - $headers = []; - foreach ($task->getHeaders() as $header) { - $pair = explode(':', $header, 2); - $headers[trim($pair[0])] = trim($pair[1]); - } - $httpRequest->setHeaders($headers); - if ($methodStr === 'POST' || $methodStr === 'PUT') { if ($task->getQueryData()) { $httpRequest->setBody(http_build_query($task->getQueryData())); From d8b2a16686d451f1e76945a8e01296698ce86b3c Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Sun, 5 Jul 2026 10:25:36 +0000 Subject: [PATCH 04/37] Fix createTask call to use positional arguments for older cloud-tasks version --- src/Api/TaskQueue/PushQueue.php | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index f997a016..1f87e9ba 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -335,12 +335,8 @@ private function addTasksCloudTasks($tasks) { $ctTask->setScheduleTime($scheduleTime); } - $request = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) - ->setParent($queueName) - ->setTask($ctTask); - try { - $response = $client->createTask($request); + $response = $client->createTask($queueName, $ctTask); $fullName = $response->getName(); $parts = explode('/', $fullName); $names[] = end($parts); From 2227b1f1f691c2df57a41b48d9891769ced130ec Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Sun, 5 Jul 2026 10:28:45 +0000 Subject: [PATCH 05/37] Enforce legacy 100KB task size limit in Cloud Tasks routing --- src/Api/TaskQueue/PushQueue.php | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 1f87e9ba..2508f7a5 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -323,7 +323,12 @@ private function addTasksCloudTasks($tasks) { if ($methodStr === 'POST' || $methodStr === 'PUT') { if ($task->getQueryData()) { - $httpRequest->setBody(http_build_query($task->getQueryData())); + $body = http_build_query($task->getQueryData()); + if (strlen($body) > PushTask::MAX_TASK_SIZE_BYTES) { + throw new TaskQueueException('Task greater than maximum size of ' . + PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); + } + $httpRequest->setBody($body); } } From dd7014ea96c8afa0c9014dc84aca3370edee1405 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 6 Jul 2026 05:47:58 +0000 Subject: [PATCH 06/37] Convert hostname to dot-notation for Cloud Tasks HTTP target HTTPS URLs --- src/Api/TaskQueue/PushQueue.php | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 2508f7a5..da7e10d5 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -307,6 +307,7 @@ private function addTasksCloudTasks($tasks) { $url = $task->getUrl(); if (strncmp($url, '/', 1) === 0) { $hostname = $hostHeader ?: \Google\AppEngine\Api\Modules\ModulesService::getHostname(); + $hostname = self::convertToDotNotation($hostname, $projectId); $url = "https://" . $hostname . $url; } $httpRequest->setUrl($url); @@ -356,4 +357,15 @@ private function addTasksCloudTasks($tasks) { } return $names; } + + private static function convertToDotNotation($hostname, $projectId) { + $parts = explode('.', $hostname); + $projectIdx = array_search($projectId, $parts); + if ($projectIdx !== false && $projectIdx > 0) { + $group1 = array_slice($parts, 0, $projectIdx + 1); + $group2 = array_slice($parts, $projectIdx + 1); + return implode('-dot-', $group1) . '.' . implode('.', $group2); + } + return $hostname; + } } From b941478235d0daea0fa95afa7f56f67057e43b8e Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Tue, 7 Jul 2026 11:26:47 +0000 Subject: [PATCH 07/37] Implement BatchCreateTasks with SDK chunking and token resolution in PHP TaskQueue diversion SDK --- src/Api/TaskQueue/PushQueue.php | 177 +++++++++++++++++++++----------- 1 file changed, 118 insertions(+), 59 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index da7e10d5..588712a9 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -275,86 +275,145 @@ private static function getProjectId() { return $projectId; } + private static function getCloudPlatformToken() { + try { + $res = \Google\AppEngine\Api\AppIdentity\AppIdentityService::getAccessToken('https://www.googleapis.com/auth/cloud-platform'); + if (isset($res['access_token'])) { + return $res['access_token']; + } + } catch (\Exception $e) { + // Fallback to metadata server if AppIdentityService fails + } + $json = self::getMetadataValue('instance/service-accounts/default/token'); + if ($json) { + $data = json_decode($json, true); + if (isset($data['access_token'])) { + return $data['access_token']; + } + } + throw new TaskQueueException('Failed to obtain OAuth access token for Cloud Tasks'); + } + private function addTasksCloudTasks($tasks) { - $client = new \Google\Cloud\Tasks\V2\CloudTasksClient(); $projectId = self::getProjectId(); $region = self::getRegion(); - $queueName = $client->queueName($projectId, $region, $this->name); + $token = self::getCloudPlatformToken(); + $fullQueueName = "projects/" . $projectId . "/locations/" . $region . "/queues/" . $this->name; $names = []; - foreach ($tasks as $task) { - $ctTask = new \Google\Cloud\Tasks\V2\Task(); - - if ($task->getName()) { - $ctTask->setName($client->taskName($projectId, $region, $this->name, $task->getName())); - } + $chunks = array_chunk($tasks, 100); - $httpRequest = new \Google\Cloud\Tasks\V2\HttpRequest(); - - $headers = []; - $hostHeader = null; - foreach ($task->getHeaders() as $header) { - $pair = explode(':', $header, 2); - $key = trim($pair[0]); - $val = trim($pair[1]); - $headers[$key] = $val; - if (strcasecmp($key, 'Host') === 0) { + foreach ($chunks as $chunk) { + $requests = []; + $chunkNames = []; + + foreach ($chunk as $task) { + $taskName = $task->getName(); + if (!$taskName) { + $taskName = 'task-' . sprintf('%04x%04x-%04x-%04x-%04x-%04x%04x%04x', + mt_rand(0, 0xffff), mt_rand(0, 0xffff), + mt_rand(0, 0xffff), + mt_rand(0, 0x0fff) | 0x4000, + mt_rand(0, 0x3fff) | 0x8000, + mt_rand(0, 0xffff), mt_rand(0, 0xffff), mt_rand(0, 0xffff)); + } + $chunkNames[] = $taskName; + $fullTaskName = $fullQueueName . "/tasks/" . $taskName; + + $headers = []; + $hostHeader = null; + foreach ($task->getHeaders() as $header) { + $pair = explode(':', $header, 2); + $key = trim($pair[0]); + $val = trim($pair[1]); + $headers[$key] = $val; + if (strcasecmp($key, 'Host') === 0) { $hostHeader = $val; + } + } + if (!isset($headers['Content-Type'])) { + $headers['Content-Type'] = 'application/octet-stream'; + } + if (!isset($headers['X-AppEngine-QueueName'])) { + $headers['X-AppEngine-QueueName'] = $this->name; + } + if (!isset($headers['X-AppEngine-TaskName'])) { + $headers['X-AppEngine-TaskName'] = $taskName; } - } - $httpRequest->setHeaders($headers); - $url = $task->getUrl(); - if (strncmp($url, '/', 1) === 0) { + $url = $task->getUrl(); + if (strncmp($url, '/', 1) === 0) { $hostname = $hostHeader ?: \Google\AppEngine\Api\Modules\ModulesService::getHostname(); $hostname = self::convertToDotNotation($hostname, $projectId); $url = "https://" . $hostname . $url; - } - $httpRequest->setUrl($url); - - $methodStr = $task->getMethod(); - $methodMap = [ - 'POST' => \Google\Cloud\Tasks\V2\HttpMethod::POST, - 'GET' => \Google\Cloud\Tasks\V2\HttpMethod::GET, - 'HEAD' => \Google\Cloud\Tasks\V2\HttpMethod::HEAD, - 'PUT' => \Google\Cloud\Tasks\V2\HttpMethod::PUT, - 'DELETE' => \Google\Cloud\Tasks\V2\HttpMethod::DELETE, - ]; - $httpRequest->setHttpMethod($methodMap[$methodStr]); + } + + $httpReq = [ + 'httpMethod' => $task->getMethod(), + 'url' => $url, + 'headers' => $headers, + ]; - if ($methodStr === 'POST' || $methodStr === 'PUT') { + if ($task->getMethod() === 'POST' || $task->getMethod() === 'PUT') { if ($task->getQueryData()) { - $body = http_build_query($task->getQueryData()); - if (strlen($body) > PushTask::MAX_TASK_SIZE_BYTES) { - throw new TaskQueueException('Task greater than maximum size of ' . - PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); - } - $httpRequest->setBody($body); + $body = http_build_query($task->getQueryData()); + if (strlen($body) > PushTask::MAX_TASK_SIZE_BYTES) { + throw new TaskQueueException('Task greater than maximum size of ' . + PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); + } + $httpReq['body'] = base64_encode($body); } + } + + $taskMap = [ + 'name' => $fullTaskName, + 'httpRequest' => $httpReq, + ]; + + if ($task->getDelaySeconds() > 0) { + $taskMap['scheduleTime'] = gmdate('Y-m-d\TH:i:s.000\Z', time() + $task->getDelaySeconds()); + } + + $requests[] = [ + 'parent' => $fullQueueName, + 'task' => $taskMap, + ]; } - - $ctTask->setHttpRequest($httpRequest); - if ($task->getDelaySeconds() > 0) { - $scheduleTime = new \Google\Protobuf\Timestamp(); - $scheduleTime->setSeconds(time() + $task->getDelaySeconds()); - $ctTask->setScheduleTime($scheduleTime); + $batchPayload = json_encode(['requests' => $requests]); + $url = "https://cloudtasks.googleapis.com/v2beta3/" . $fullQueueName . "/tasks:batchCreate"; + + $opts = [ + 'http' => [ + 'method' => 'POST', + 'header' => "Authorization: Bearer " . $token . "\r\n" . + "Content-Type: application/json\r\n", + 'content' => $batchPayload, + 'timeout' => 10.0, + 'ignore_errors' => true, + ] + ]; + $context = stream_context_create($opts); + $response = @file_get_contents($url, false, $context); + + $statusLine = $http_response_header[0] ?? ''; + if (preg_match('#HTTP/\d+\.\d+\s+([0-9]{3})#', $statusLine, $matches)) { + $code = intval($matches[1]); + } else { + $code = 500; } - try { - $response = $client->createTask($queueName, $ctTask); - $fullName = $response->getName(); - $parts = explode('/', $fullName); - $names[] = end($parts); - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS') { - throw new TaskAlreadyExistsException('Task with the same name exists already'); - } - throw new TaskQueueException('Cloud Tasks Error: ' . $e->getMessage(), $e->getCode()); - } catch (\Exception $e) { - throw new TaskQueueException('Error calling Cloud Tasks: ' . $e->getMessage()); + if ($code === 200 || $code === 201) { + foreach ($chunkNames as $name) { + $names[] = $name; + } + } else if ($code === 409) { + throw new TaskAlreadyExistsException('Task with the same name exists already'); + } else { + throw new TaskQueueException('Cloud Tasks batchCreate failed with status ' . $code . ': ' . $response); } } + return $names; } From ed31320ff2291bd978ebbd769e35642d60b60df2 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Thu, 9 Jul 2026 06:19:51 +0000 Subject: [PATCH 08/37] Fix AppIdentityService APC/Memcache calls for PHP 8.2 compatibility --- src/Api/AppIdentity/AppIdentityService.php | 37 +++++++++++++++------- 1 file changed, 26 insertions(+), 11 deletions(-) diff --git a/src/Api/AppIdentity/AppIdentityService.php b/src/Api/AppIdentity/AppIdentityService.php index 077d3bb2..747bd73c 100644 --- a/src/Api/AppIdentity/AppIdentityService.php +++ b/src/Api/AppIdentity/AppIdentityService.php @@ -274,8 +274,12 @@ public static function getDefaultVersionHostname() { private static function putTokenInCache($name, $value, $expiry_secs) { $expiry_time_from_epoch = $expiry_secs - self::EXPIRY_SAFETY_MARGIN_SECS - self::EXPIRY_SHORT_MARGIN_SECS; - $memcache = new Memcache(); - $memcache->set($name, $value, null, $expiry_time_from_epoch); + if (class_exists('Memcache')) { + try { + $memcache = new Memcache(); + $memcache->set($name, $value, null, $expiry_time_from_epoch); + } catch (\Throwable $t) {} + } // Record the expiry time in the object being cached, so we can check it // when read from APC. self::putTokenInApc($name, $value, $expiry_secs); @@ -296,7 +300,11 @@ private static function putTokenInApc($name, $value, $expiry_secs) { self::EXPIRY_SHORT_MARGIN_SECS; $cache_ttl = self::getTTLForToken($expiry_time_from_epoch); $value['eviction_time_epoch'] = $cache_ttl['eviction_time_epoch']; - apc_store($name, $value, $cache_ttl['apc_ttl_in_seconds']); + if (function_exists('apcu_store')) { + apcu_store($name, $value, $cache_ttl['apc_ttl_in_seconds']); + } elseif (function_exists('apc_store')) { + apc_store($name, $value, $cache_ttl['apc_ttl_in_seconds']); + } } /** @@ -311,17 +319,24 @@ private static function putTokenInApc($name, $value, $expiry_secs) { */ private static function getTokenFromCache($name) { $success = false; - $result = apc_fetch($name, $success); - if ($success && time() < $result['eviction_time_epoch']) { + $result = false; + if (function_exists('apcu_fetch')) { + $result = apcu_fetch($name, $success); + } elseif (function_exists('apc_fetch')) { + $result = apc_fetch($name, $success); + } + if ($success && $result !== false && time() < $result['eviction_time_epoch']) { unset($result['eviction_time_epoch']); return $result; } - $memcache = new Memcache(); - $result = $memcache->get($name); - // If there was a result in memcache but not in apc we can add using a - // short timeout. - if ($result !== false) { - self::putTokenInApc($name, $result, $result['expiration_time']); + if (class_exists('Memcache')) { + try { + $memcache = new Memcache(); + $result = $memcache->get($name); + if ($result !== false) { + self::putTokenInApc($name, $result, $result['expiration_time']); + } + } catch (\Throwable $t) {} } return $result; } From d42d91d20e53c3415c06f79867fe26f4f6bd2411 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Thu, 9 Jul 2026 06:30:28 +0000 Subject: [PATCH 09/37] Parse individual task error status in batchCreate response for Cloud Tasks routing --- src/Api/TaskQueue/PushQueue.php | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 588712a9..2d5d983f 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -404,6 +404,20 @@ private function addTasksCloudTasks($tasks) { } if ($code === 200 || $code === 201) { + $resData = json_decode($response, true); + if (isset($resData['tasks']) && is_array($resData['tasks'])) { + foreach ($resData['tasks'] as $idx => $item) { + if (isset($item['status']) && isset($item['status']['code']) && (int)$item['status']['code'] !== 0) { + $statusCode = (int)$item['status']['code']; + $statusMsg = $item['status']['message'] ?? 'Unknown error in batchCreate'; + if ($statusCode === 6 || $statusCode === 409 || stripos($statusMsg, 'already exists') !== false) { + throw new TaskAlreadyExistsException('Task with the same name exists already: ' . $statusMsg); + } else { + throw new TaskQueueException('Cloud Tasks batchCreate task failed (' . $statusCode . '): ' . $statusMsg); + } + } + } + } foreach ($chunkNames as $name) { $names[] = $name; } From 2a7d64b06d79042b84d2b2bd0276c945bb3830f4 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Thu, 9 Jul 2026 07:02:12 +0000 Subject: [PATCH 10/37] Fix ALREADY_EXISTS and NOT_FOUND error handling in Cloud Tasks batchCreate response --- src/Api/TaskQueue/PushQueue.php | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 2d5d983f..d84c534f 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -401,8 +401,6 @@ private function addTasksCloudTasks($tasks) { $code = intval($matches[1]); } else { $code = 500; - } - if ($code === 200 || $code === 201) { $resData = json_decode($response, true); if (isset($resData['tasks']) && is_array($resData['tasks'])) { From 0e0819633b079e0638a266fb2cc675c0cefcb35e Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Thu, 9 Jul 2026 07:06:16 +0000 Subject: [PATCH 11/37] Fix syntax error (missing closing brace) in PushQueue.php --- src/Api/TaskQueue/PushQueue.php | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index d84c534f..1b14e832 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -401,6 +401,7 @@ private function addTasksCloudTasks($tasks) { $code = intval($matches[1]); } else { $code = 500; + } if ($code === 200 || $code === 201) { $resData = json_decode($response, true); if (isset($resData['tasks']) && is_array($resData['tasks'])) { From 200e51b578d3f9041981c913910f3c2987913e79 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Thu, 9 Jul 2026 07:14:32 +0000 Subject: [PATCH 12/37] Check Operation error and metadata.failedRequests in Cloud Tasks batchCreate response --- src/Api/TaskQueue/PushQueue.php | 26 ++++++++++++++++++-------- 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 1b14e832..f250a9e3 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -404,15 +404,25 @@ private function addTasksCloudTasks($tasks) { } if ($code === 200 || $code === 201) { $resData = json_decode($response, true); - if (isset($resData['tasks']) && is_array($resData['tasks'])) { - foreach ($resData['tasks'] as $idx => $item) { - if (isset($item['status']) && isset($item['status']['code']) && (int)$item['status']['code'] !== 0) { - $statusCode = (int)$item['status']['code']; - $statusMsg = $item['status']['message'] ?? 'Unknown error in batchCreate'; - if ($statusCode === 6 || $statusCode === 409 || stripos($statusMsg, 'already exists') !== false) { - throw new TaskAlreadyExistsException('Task with the same name exists already: ' . $statusMsg); + if (is_array($resData)) { + if (isset($resData['error']) && isset($resData['error']['code']) && (int)$resData['error']['code'] !== 0) { + $errCode = (int)$resData['error']['code']; + $errMsg = $resData['error']['message'] ?? 'BatchCreateTasks operation failed'; + if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false) { + throw new TaskAlreadyExistsException('Task exists already: ' . $errMsg); + } else { + throw new TaskQueueException('Cloud Tasks batchCreate failed (' . $errCode . '): ' . $errMsg); + } + } + $failedReqs = $resData['metadata']['failedRequests'] ?? ($resData['metadata']['failed_requests'] ?? null); + if (is_array($failedReqs)) { + foreach ($failedReqs as $idx => $err) { + $errCode = (int)($err['code'] ?? 0); + $errMsg = $err['message'] ?? 'Task creation failed'; + if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false) { + throw new TaskAlreadyExistsException('Task exists already: ' . $errMsg); } else { - throw new TaskQueueException('Cloud Tasks batchCreate task failed (' . $statusCode . '): ' . $statusMsg); + throw new TaskQueueException('Cloud Tasks batchCreate task failed (' . $errCode . '): ' . $errMsg); } } } From be5dc7b1e6e220982f7dbd601af0226eaa1709e3 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Thu, 9 Jul 2026 07:17:31 +0000 Subject: [PATCH 13/37] Normalize HTTP header casing in PushQueue to prevent duplicate Content-Type error --- src/Api/TaskQueue/PushQueue.php | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index f250a9e3..e3d26ff8 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -322,16 +322,25 @@ private function addTasksCloudTasks($tasks) { $headers = []; $hostHeader = null; + $hasContentType = false; foreach ($task->getHeaders() as $header) { $pair = explode(':', $header, 2); $key = trim($pair[0]); $val = trim($pair[1]); - $headers[$key] = $val; if (strcasecmp($key, 'Host') === 0) { $hostHeader = $val; + $key = 'Host'; + } elseif (strcasecmp($key, 'Content-Type') === 0) { + $hasContentType = true; + $key = 'Content-Type'; + } elseif (strcasecmp($key, 'X-AppEngine-QueueName') === 0) { + $key = 'X-AppEngine-QueueName'; + } elseif (strcasecmp($key, 'X-AppEngine-TaskName') === 0) { + $key = 'X-AppEngine-TaskName'; } + $headers[$key] = $val; } - if (!isset($headers['Content-Type'])) { + if (!$hasContentType) { $headers['Content-Type'] = 'application/octet-stream'; } if (!isset($headers['X-AppEngine-QueueName'])) { From f20bbf8fdb65f4336b3d21cc5fab53ef134b5418 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Thu, 9 Jul 2026 07:20:15 +0000 Subject: [PATCH 14/37] Map 404/NOT_FOUND (Requested entity was not found) on task creation to TaskAlreadyExistsException for tombstoned tasks --- src/Api/TaskQueue/PushQueue.php | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index e3d26ff8..fa02babc 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -417,8 +417,8 @@ private function addTasksCloudTasks($tasks) { if (isset($resData['error']) && isset($resData['error']['code']) && (int)$resData['error']['code'] !== 0) { $errCode = (int)$resData['error']['code']; $errMsg = $resData['error']['message'] ?? 'BatchCreateTasks operation failed'; - if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false) { - throw new TaskAlreadyExistsException('Task exists already: ' . $errMsg); + if (self::isAlreadyExistsError($errCode, $errMsg)) { + throw new TaskAlreadyExistsException('Task exists already (or is tombstoned): ' . $errMsg); } else { throw new TaskQueueException('Cloud Tasks batchCreate failed (' . $errCode . '): ' . $errMsg); } @@ -428,8 +428,8 @@ private function addTasksCloudTasks($tasks) { foreach ($failedReqs as $idx => $err) { $errCode = (int)($err['code'] ?? 0); $errMsg = $err['message'] ?? 'Task creation failed'; - if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false) { - throw new TaskAlreadyExistsException('Task exists already: ' . $errMsg); + if (self::isAlreadyExistsError($errCode, $errMsg)) { + throw new TaskAlreadyExistsException('Task exists already (or is tombstoned): ' . $errMsg); } else { throw new TaskQueueException('Cloud Tasks batchCreate task failed (' . $errCode . '): ' . $errMsg); } @@ -439,8 +439,8 @@ private function addTasksCloudTasks($tasks) { foreach ($chunkNames as $name) { $names[] = $name; } - } else if ($code === 409) { - throw new TaskAlreadyExistsException('Task with the same name exists already'); + } else if ($code === 409 || self::isAlreadyExistsError($code, $response)) { + throw new TaskAlreadyExistsException('Task with the same name exists already (or is tombstoned)'); } else { throw new TaskQueueException('Cloud Tasks batchCreate failed with status ' . $code . ': ' . $response); } @@ -449,6 +449,16 @@ private function addTasksCloudTasks($tasks) { return $names; } + private static function isAlreadyExistsError($errCode, $errMsg) { + if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false) { + return true; + } + if (($errCode === 5 || $errCode === 404) && stripos($errMsg, 'Requested entity was not found') !== false) { + return true; + } + return false; + } + private static function convertToDotNotation($hostname, $projectId) { $parts = explode('.', $hostname); $projectIdx = array_search($projectId, $parts); From 2184c803ac3cb3b0a2eeb31b152970ddd65db4e8 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 20 Jul 2026 10:09:58 +0000 Subject: [PATCH 15/37] Fix: Strip version ID from hostname to route tasks to active service endpoint --- src/Api/TaskQueue/PushQueue.php | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index fa02babc..a486b960 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -465,6 +465,9 @@ private static function convertToDotNotation($hostname, $projectId) { if ($projectIdx !== false && $projectIdx > 0) { $group1 = array_slice($parts, 0, $projectIdx + 1); $group2 = array_slice($parts, $projectIdx + 1); + if (count($group1) > 2) { + array_shift($group1); + } return implode('-dot-', $group1) . '.' . implode('.', $group2); } return $hostname; From 8d34bd18c7e8e6aeb321c2ef7f0f2b1d37c166f4 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Fri, 31 Jul 2026 08:35:26 +0000 Subject: [PATCH 16/37] Use CloudTasks Client SDK for batchCreateTasks on dogfood branch --- src/Api/TaskQueue/PushQueue.php | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index a486b960..b280d413 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -392,6 +392,7 @@ private function addTasksCloudTasks($tasks) { $batchPayload = json_encode(['requests' => $requests]); $url = "https://cloudtasks.googleapis.com/v2beta3/" . $fullQueueName . "/tasks:batchCreate"; + // On dogfood branch, use Client SDK for BatchCreateTasks $opts = [ 'http' => [ 'method' => 'POST', From 9eb55425997387a2d8b202ed5437d77be76641e5 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Sat, 1 Aug 2026 10:43:49 +0000 Subject: [PATCH 17/37] Update PushQueue to use Google Cloud Tasks V2beta3 CloudTasksClient --- src/Api/TaskQueue/PushQueue.php | 74 ++++++++++----------------------- 1 file changed, 22 insertions(+), 52 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index b280d413..4f5cafed 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -389,61 +389,31 @@ private function addTasksCloudTasks($tasks) { ]; } - $batchPayload = json_encode(['requests' => $requests]); - $url = "https://cloudtasks.googleapis.com/v2beta3/" . $fullQueueName . "/tasks:batchCreate"; + // On dogfood branch, use Client SDK for Task creation + $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); + try { + foreach ($chunk as $idx => $task) { + $tName = $chunkNames[$idx]; + $fTaskName = "projects/" . $projectId . "/locations/" . $region . "/queues/" . $this->name . "/tasks/" . $tName; - // On dogfood branch, use Client SDK for BatchCreateTasks - $opts = [ - 'http' => [ - 'method' => 'POST', - 'header' => "Authorization: Bearer " . $token . "\r\n" . - "Content-Type: application/json\r\n", - 'content' => $batchPayload, - 'timeout' => 10.0, - 'ignore_errors' => true, - ] - ]; - $context = stream_context_create($opts); - $response = @file_get_contents($url, false, $context); - - $statusLine = $http_response_header[0] ?? ''; - if (preg_match('#HTTP/\d+\.\d+\s+([0-9]{3})#', $statusLine, $matches)) { - $code = intval($matches[1]); - } else { - $code = 500; - } - if ($code === 200 || $code === 201) { - $resData = json_decode($response, true); - if (is_array($resData)) { - if (isset($resData['error']) && isset($resData['error']['code']) && (int)$resData['error']['code'] !== 0) { - $errCode = (int)$resData['error']['code']; - $errMsg = $resData['error']['message'] ?? 'BatchCreateTasks operation failed'; - if (self::isAlreadyExistsError($errCode, $errMsg)) { - throw new TaskAlreadyExistsException('Task exists already (or is tombstoned): ' . $errMsg); - } else { - throw new TaskQueueException('Cloud Tasks batchCreate failed (' . $errCode . '): ' . $errMsg); - } - } - $failedReqs = $resData['metadata']['failedRequests'] ?? ($resData['metadata']['failed_requests'] ?? null); - if (is_array($failedReqs)) { - foreach ($failedReqs as $idx => $err) { - $errCode = (int)($err['code'] ?? 0); - $errMsg = $err['message'] ?? 'Task creation failed'; - if (self::isAlreadyExistsError($errCode, $errMsg)) { - throw new TaskAlreadyExistsException('Task exists already (or is tombstoned): ' . $errMsg); - } else { - throw new TaskQueueException('Cloud Tasks batchCreate task failed (' . $errCode . '): ' . $errMsg); - } - } - } + $httpReq = new \Google\Cloud\Tasks\V2beta3\HttpRequest(); + $httpReq->setUrl($requests[$idx]['task']['httpRequest']['url']); + $httpReq->setHttpMethod(\Google\Cloud\Tasks\V2beta3\HttpMethod::POST); + + $taskObj = new \Google\Cloud\Tasks\V2beta3\Task(); + $taskObj->setName($fTaskName); + $taskObj->setHttpRequest($httpReq); + + $client->createTask($fullQueueName, $taskObj); + $names[] = $tName; } - foreach ($chunkNames as $name) { - $names[] = $name; + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); } - } else if ($code === 409 || self::isAlreadyExistsError($code, $response)) { - throw new TaskAlreadyExistsException('Task with the same name exists already (or is tombstoned)'); - } else { - throw new TaskQueueException('Cloud Tasks batchCreate failed with status ' . $code . ': ' . $response); + throw new TaskQueueException('Cloud Tasks Client SDK creation failed: ' . $e->getMessage()); + } finally { + $client->close(); } } From c6695c1165bea03ee077f4cd937253fc2b72912b Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Sat, 1 Aug 2026 13:53:59 +0000 Subject: [PATCH 18/37] Use batchCreateTasks REST endpoint for batch task creation --- src/Api/TaskQueue/PushQueue.php | 74 +++++++++++++++++++++++---------- 1 file changed, 52 insertions(+), 22 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 4f5cafed..c8cc6221 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -389,31 +389,61 @@ private function addTasksCloudTasks($tasks) { ]; } - // On dogfood branch, use Client SDK for Task creation - $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); - try { - foreach ($chunk as $idx => $task) { - $tName = $chunkNames[$idx]; - $fTaskName = "projects/" . $projectId . "/locations/" . $region . "/queues/" . $this->name . "/tasks/" . $tName; - - $httpReq = new \Google\Cloud\Tasks\V2beta3\HttpRequest(); - $httpReq->setUrl($requests[$idx]['task']['httpRequest']['url']); - $httpReq->setHttpMethod(\Google\Cloud\Tasks\V2beta3\HttpMethod::POST); - - $taskObj = new \Google\Cloud\Tasks\V2beta3\Task(); - $taskObj->setName($fTaskName); - $taskObj->setHttpRequest($httpReq); + $batchPayload = json_encode(['requests' => $requests]); + $url = "https://cloudtasks.googleapis.com/v2beta3/" . $fullQueueName . "/tasks:batchCreate"; - $client->createTask($fullQueueName, $taskObj); - $names[] = $tName; + // On dogfood branch, call BatchCreateTasks RPC via v2beta3 REST endpoint + $opts = [ + 'http' => [ + 'method' => 'POST', + 'header' => "Authorization: Bearer " . $token . "\r\n" . + "Content-Type: application/json\r\n", + 'content' => $batchPayload, + 'timeout' => 10.0, + 'ignore_errors' => true, + ] + ]; + $context = stream_context_create($opts); + $response = @file_get_contents($url, false, $context); + + $statusLine = $http_response_header[0] ?? ''; + if (preg_match('#HTTP/\d+\.\d+\s+([0-9]{3})#', $statusLine, $matches)) { + $code = intval($matches[1]); + } else { + $code = 500; + } + if ($code === 200 || $code === 201) { + $resData = json_decode($response, true); + if (is_array($resData)) { + if (isset($resData['error']) && isset($resData['error']['code']) && (int)$resData['error']['code'] !== 0) { + $errCode = (int)$resData['error']['code']; + $errMsg = $resData['error']['message'] ?? 'BatchCreateTasks operation failed'; + if (self::isAlreadyExistsError($errCode, $errMsg)) { + throw new TaskAlreadyExistsException('Task exists already (or is tombstoned): ' . $errMsg); + } else { + throw new TaskQueueException('Cloud Tasks batchCreate failed (' . $errCode . '): ' . $errMsg); + } + } + $failedReqs = $resData['metadata']['failedRequests'] ?? ($resData['metadata']['failed_requests'] ?? null); + if (is_array($failedReqs)) { + foreach ($failedReqs as $idx => $err) { + $errCode = (int)($err['code'] ?? 0); + $errMsg = $err['message'] ?? 'Task creation failed'; + if (self::isAlreadyExistsError($errCode, $errMsg)) { + throw new TaskAlreadyExistsException('Task exists already (or is tombstoned): ' . $errMsg); + } else { + throw new TaskQueueException('Cloud Tasks batchCreate task failed (' . $errCode . '): ' . $errMsg); + } + } + } } - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + foreach ($chunkNames as $name) { + $names[] = $name; } - throw new TaskQueueException('Cloud Tasks Client SDK creation failed: ' . $e->getMessage()); - } finally { - $client->close(); + } else if ($code === 409 || self::isAlreadyExistsError($code, $response)) { + throw new TaskAlreadyExistsException('Task with the same name exists already (or is tombstoned)'); + } else { + throw new TaskQueueException('Cloud Tasks batchCreate failed with status ' . $code . ': ' . $response); } } From ea2f1322d4478e26a7641ed6c424bdb6d4cc54f2 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Sat, 1 Aug 2026 14:23:05 +0000 Subject: [PATCH 19/37] Use CloudTasksClient batchCreateTasks in PushQueue.php --- src/Api/TaskQueue/PushQueue.php | 80 ++++++++++++--------------------- 1 file changed, 29 insertions(+), 51 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index c8cc6221..7aad65ad 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -389,61 +389,39 @@ private function addTasksCloudTasks($tasks) { ]; } - $batchPayload = json_encode(['requests' => $requests]); - $url = "https://cloudtasks.googleapis.com/v2beta3/" . $fullQueueName . "/tasks:batchCreate"; - - // On dogfood branch, call BatchCreateTasks RPC via v2beta3 REST endpoint - $opts = [ - 'http' => [ - 'method' => 'POST', - 'header' => "Authorization: Bearer " . $token . "\r\n" . - "Content-Type: application/json\r\n", - 'content' => $batchPayload, - 'timeout' => 10.0, - 'ignore_errors' => true, - ] - ]; - $context = stream_context_create($opts); - $response = @file_get_contents($url, false, $context); - - $statusLine = $http_response_header[0] ?? ''; - if (preg_match('#HTTP/\d+\.\d+\s+([0-9]{3})#', $statusLine, $matches)) { - $code = intval($matches[1]); - } else { - $code = 500; - } - if ($code === 200 || $code === 201) { - $resData = json_decode($response, true); - if (is_array($resData)) { - if (isset($resData['error']) && isset($resData['error']['code']) && (int)$resData['error']['code'] !== 0) { - $errCode = (int)$resData['error']['code']; - $errMsg = $resData['error']['message'] ?? 'BatchCreateTasks operation failed'; - if (self::isAlreadyExistsError($errCode, $errMsg)) { - throw new TaskAlreadyExistsException('Task exists already (or is tombstoned): ' . $errMsg); - } else { - throw new TaskQueueException('Cloud Tasks batchCreate failed (' . $errCode . '): ' . $errMsg); - } - } - $failedReqs = $resData['metadata']['failedRequests'] ?? ($resData['metadata']['failed_requests'] ?? null); - if (is_array($failedReqs)) { - foreach ($failedReqs as $idx => $err) { - $errCode = (int)($err['code'] ?? 0); - $errMsg = $err['message'] ?? 'Task creation failed'; - if (self::isAlreadyExistsError($errCode, $errMsg)) { - throw new TaskAlreadyExistsException('Task exists already (or is tombstoned): ' . $errMsg); - } else { - throw new TaskQueueException('Cloud Tasks batchCreate task failed (' . $errCode . '): ' . $errMsg); - } - } - } + // On dogfood branch, use Client SDK for batchCreateTasks + $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); + try { + $createTaskRequests = []; + foreach ($chunk as $idx => $task) { + $tName = $chunkNames[$idx]; + $fTaskName = "projects/" . $projectId . "/locations/" . $region . "/queues/" . $this->name . "/tasks/" . $tName; + + $httpReq = new \Google\Cloud\Tasks\V2beta3\HttpRequest(); + $httpReq->setUrl($requests[$idx]['task']['httpRequest']['url']); + $httpReq->setHttpMethod(\Google\Cloud\Tasks\V2beta3\HttpMethod::POST); + + $taskObj = new \Google\Cloud\Tasks\V2beta3\Task(); + $taskObj->setName($fTaskName); + $taskObj->setHttpRequest($httpReq); + + $createTaskReq = new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest(); + $createTaskReq->setParent($fullQueueName); + $createTaskReq->setTask($taskObj); + $createTaskRequests[] = $createTaskReq; } + + $client->batchCreateTasks($fullQueueName, $createTaskRequests); foreach ($chunkNames as $name) { $names[] = $name; } - } else if ($code === 409 || self::isAlreadyExistsError($code, $response)) { - throw new TaskAlreadyExistsException('Task with the same name exists already (or is tombstoned)'); - } else { - throw new TaskQueueException('Cloud Tasks batchCreate failed with status ' . $code . ': ' . $response); + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK batchCreate failed: ' . $e->getMessage()); + } finally { + $client->close(); } } From f2960e9b59641703d948eb02511aa70aee05d883 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 3 Aug 2026 07:25:47 +0000 Subject: [PATCH 20/37] Update region detection in PushQueue.php to match Python SDK and revert PushTask.php whitespace --- src/Api/TaskQueue/PushQueue.php | 18 ++++++------------ src/Api/TaskQueue/PushTask.php | 2 +- 2 files changed, 7 insertions(+), 13 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 7aad65ad..6bb96b2d 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -235,22 +235,16 @@ private static function getMetadataValue($path) { private static function getRegion() { static $region = null; if ($region === null) { - $region = getenv('REGION_ID'); + $region = getenv('LOCATION_ID') ?: getenv('GAE_LOCATION') ?: getenv('GAE_REGION') ?: getenv('REGION_ID'); if (!$region) { - $zone = self::getMetadataValue('instance/zone'); - if ($zone) { - $parts = explode('/', $zone); - $zoneName = end($parts); - $dashPos = strrpos($zoneName, '-'); - if ($dashPos !== false) { - $region = substr($zoneName, 0, $dashPos); - } else { - $region = $zoneName; - } + $regionPath = self::getMetadataValue('instance/region'); + if ($regionPath) { + $parts = explode('/', $regionPath); + $region = end($parts); } } if (!$region) { - $region = 'us-central1'; + $region = getenv('LOCAL_GCP_REGION') ?: 'us-central1'; } } return $region; diff --git a/src/Api/TaskQueue/PushTask.php b/src/Api/TaskQueue/PushTask.php index 96d03248..e0409810 100644 --- a/src/Api/TaskQueue/PushTask.php +++ b/src/Api/TaskQueue/PushTask.php @@ -287,7 +287,7 @@ public function getHeaders() { * exists in the queue. * @throws TaskQueueException if there was a problem using the service. */ - public function add($queue_name = 'default') { + public function add($queue_name = 'default') { $queue = new PushQueue($queue_name); return $queue->addTasks([$this])[0]; } From d79fe594f303ec0d17d6a4d0f1f416f592ad8d3d Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 3 Aug 2026 07:33:53 +0000 Subject: [PATCH 21/37] Clean up addTasksCloudTasks in PushQueue.php: use server-generated task names and remove dogfood comment --- src/Api/TaskQueue/PushQueue.php | 82 ++++++++++++--------------------- 1 file changed, 29 insertions(+), 53 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 6bb96b2d..6141927c 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -298,22 +298,9 @@ private function addTasksCloudTasks($tasks) { $chunks = array_chunk($tasks, 100); foreach ($chunks as $chunk) { - $requests = []; - $chunkNames = []; + $createTaskRequests = []; foreach ($chunk as $task) { - $taskName = $task->getName(); - if (!$taskName) { - $taskName = 'task-' . sprintf('%04x%04x-%04x-%04x-%04x-%04x%04x%04x', - mt_rand(0, 0xffff), mt_rand(0, 0xffff), - mt_rand(0, 0xffff), - mt_rand(0, 0x0fff) | 0x4000, - mt_rand(0, 0x3fff) | 0x8000, - mt_rand(0, 0xffff), mt_rand(0, 0xffff), mt_rand(0, 0xffff)); - } - $chunkNames[] = $taskName; - $fullTaskName = $fullQueueName . "/tasks/" . $taskName; - $headers = []; $hostHeader = null; $hasContentType = false; @@ -340,7 +327,9 @@ private function addTasksCloudTasks($tasks) { if (!isset($headers['X-AppEngine-QueueName'])) { $headers['X-AppEngine-QueueName'] = $this->name; } - if (!isset($headers['X-AppEngine-TaskName'])) { + + $taskName = $task->getName(); + if ($taskName && !isset($headers['X-AppEngine-TaskName'])) { $headers['X-AppEngine-TaskName'] = $taskName; } @@ -351,11 +340,13 @@ private function addTasksCloudTasks($tasks) { $url = "https://" . $hostname . $url; } - $httpReq = [ - 'httpMethod' => $task->getMethod(), - 'url' => $url, - 'headers' => $headers, - ]; + $httpReq = new \Google\Cloud\Tasks\V2beta3\HttpRequest(); + $httpReq->setUrl($url); + $httpReq->setHttpMethod(\Google\Cloud\Tasks\V2beta3\HttpMethod::POST); + + foreach ($headers as $k => $v) { + $httpReq->getHeaders()[$k] = $v; + } if ($task->getMethod() === 'POST' || $task->getMethod() === 'PUT') { if ($task->getQueryData()) { @@ -364,50 +355,35 @@ private function addTasksCloudTasks($tasks) { throw new TaskQueueException('Task greater than maximum size of ' . PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); } - $httpReq['body'] = base64_encode($body); + $httpReq->setBody($body); } } - $taskMap = [ - 'name' => $fullTaskName, - 'httpRequest' => $httpReq, - ]; + $taskObj = new \Google\Cloud\Tasks\V2beta3\Task(); + if ($taskName) { + $fullTaskName = $fullQueueName . "/tasks/" . $taskName; + $taskObj->setName($fullTaskName); + } + $taskObj->setHttpRequest($httpReq); if ($task->getDelaySeconds() > 0) { - $taskMap['scheduleTime'] = gmdate('Y-m-d\TH:i:s.000\Z', time() + $task->getDelaySeconds()); + $ts = new \Google\Protobuf\Timestamp(); + $ts->setSeconds(time() + $task->getDelaySeconds()); + $taskObj->setScheduleTime($ts); } - $requests[] = [ - 'parent' => $fullQueueName, - 'task' => $taskMap, - ]; + $createTaskReq = new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest(); + $createTaskReq->setParent($fullQueueName); + $createTaskReq->setTask($taskObj); + $createTaskRequests[] = $createTaskReq; } - // On dogfood branch, use Client SDK for batchCreateTasks $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); try { - $createTaskRequests = []; - foreach ($chunk as $idx => $task) { - $tName = $chunkNames[$idx]; - $fTaskName = "projects/" . $projectId . "/locations/" . $region . "/queues/" . $this->name . "/tasks/" . $tName; - - $httpReq = new \Google\Cloud\Tasks\V2beta3\HttpRequest(); - $httpReq->setUrl($requests[$idx]['task']['httpRequest']['url']); - $httpReq->setHttpMethod(\Google\Cloud\Tasks\V2beta3\HttpMethod::POST); - - $taskObj = new \Google\Cloud\Tasks\V2beta3\Task(); - $taskObj->setName($fTaskName); - $taskObj->setHttpRequest($httpReq); - - $createTaskReq = new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest(); - $createTaskReq->setParent($fullQueueName); - $createTaskReq->setTask($taskObj); - $createTaskRequests[] = $createTaskReq; - } - - $client->batchCreateTasks($fullQueueName, $createTaskRequests); - foreach ($chunkNames as $name) { - $names[] = $name; + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + foreach ($response->getTasks() as $resTask) { + $parts = explode('/', $resTask->getName()); + $names[] = end($parts); } } catch (\Google\ApiCore\ApiException $e) { if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409) { From c4ab5f17af9c595a66748901f2be2e980590ee14 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 3 Aug 2026 07:38:12 +0000 Subject: [PATCH 22/37] Switch PushQueue.php from HttpRequest with full URL to AppEngineHttpRequest with relativeUri, removing convertToDotNotation --- src/Api/TaskQueue/PushQueue.php | 46 +++++++++++---------------------- 1 file changed, 15 insertions(+), 31 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 6141927c..5a09ee37 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -302,16 +302,12 @@ private function addTasksCloudTasks($tasks) { foreach ($chunk as $task) { $headers = []; - $hostHeader = null; $hasContentType = false; foreach ($task->getHeaders() as $header) { $pair = explode(':', $header, 2); $key = trim($pair[0]); $val = trim($pair[1]); - if (strcasecmp($key, 'Host') === 0) { - $hostHeader = $val; - $key = 'Host'; - } elseif (strcasecmp($key, 'Content-Type') === 0) { + if (strcasecmp($key, 'Content-Type') === 0) { $hasContentType = true; $key = 'Content-Type'; } elseif (strcasecmp($key, 'X-AppEngine-QueueName') === 0) { @@ -333,19 +329,21 @@ private function addTasksCloudTasks($tasks) { $headers['X-AppEngine-TaskName'] = $taskName; } - $url = $task->getUrl(); - if (strncmp($url, '/', 1) === 0) { - $hostname = $hostHeader ?: \Google\AppEngine\Api\Modules\ModulesService::getHostname(); - $hostname = self::convertToDotNotation($hostname, $projectId); - $url = "https://" . $hostname . $url; - } + $methodMap = [ + 'POST' => \Google\Cloud\Tasks\V2beta3\HttpMethod::POST, + 'GET' => \Google\Cloud\Tasks\V2beta3\HttpMethod::GET, + 'PUT' => \Google\Cloud\Tasks\V2beta3\HttpMethod::PUT, + 'DELETE' => \Google\Cloud\Tasks\V2beta3\HttpMethod::DELETE, + 'HEAD' => \Google\Cloud\Tasks\V2beta3\HttpMethod::HEAD, + ]; + $httpMethod = isset($methodMap[$task->getMethod()]) ? $methodMap[$task->getMethod()] : \Google\Cloud\Tasks\V2beta3\HttpMethod::POST; - $httpReq = new \Google\Cloud\Tasks\V2beta3\HttpRequest(); - $httpReq->setUrl($url); - $httpReq->setHttpMethod(\Google\Cloud\Tasks\V2beta3\HttpMethod::POST); + $appEngineReq = new \Google\Cloud\Tasks\V2beta3\AppEngineHttpRequest(); + $appEngineReq->setRelativeUri($task->getUrl() ?: '/'); + $appEngineReq->setHttpMethod($httpMethod); foreach ($headers as $k => $v) { - $httpReq->getHeaders()[$k] = $v; + $appEngineReq->getHeaders()[$k] = $v; } if ($task->getMethod() === 'POST' || $task->getMethod() === 'PUT') { @@ -355,7 +353,7 @@ private function addTasksCloudTasks($tasks) { throw new TaskQueueException('Task greater than maximum size of ' . PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); } - $httpReq->setBody($body); + $appEngineReq->setBody($body); } } @@ -364,7 +362,7 @@ private function addTasksCloudTasks($tasks) { $fullTaskName = $fullQueueName . "/tasks/" . $taskName; $taskObj->setName($fullTaskName); } - $taskObj->setHttpRequest($httpReq); + $taskObj->setAppEngineHttpRequest($appEngineReq); if ($task->getDelaySeconds() > 0) { $ts = new \Google\Protobuf\Timestamp(); @@ -407,18 +405,4 @@ private static function isAlreadyExistsError($errCode, $errMsg) { } return false; } - - private static function convertToDotNotation($hostname, $projectId) { - $parts = explode('.', $hostname); - $projectIdx = array_search($projectId, $parts); - if ($projectIdx !== false && $projectIdx > 0) { - $group1 = array_slice($parts, 0, $projectIdx + 1); - $group2 = array_slice($parts, $projectIdx + 1); - if (count($group1) > 2) { - array_shift($group1); - } - return implode('-dot-', $group1) . '.' . implode('.', $group2); - } - return $hostname; - } } From 18cc91f0ffa9eb8db504f386332b441395ca086e Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 3 Aug 2026 07:42:10 +0000 Subject: [PATCH 23/37] Unpack metadata failed_requests from BatchCreateTasks OperationResponse to catch individual task creation errors --- src/Api/TaskQueue/PushQueue.php | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 5a09ee37..7ec67712 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -379,9 +379,26 @@ private function addTasksCloudTasks($tasks) { $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); try { $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); - foreach ($response->getTasks() as $resTask) { - $parts = explode('/', $resTask->getName()); - $names[] = end($parts); + $metadata = $response->getMetadata(); + if ($metadata && $metadata->getFailedRequests()) { + foreach ($chunk as $idx => $task) { + if ($metadata->getFailedRequests()->offsetExists($idx)) { + $errStatus = $metadata->getFailedRequests()->offsetGet($idx); + $code = $errStatus->getCode(); + $msg = $errStatus->getMessage(); + if (self::isAlreadyExistsError($code, $msg)) { + throw new TaskAlreadyExistsException('Task exists already: ' . $msg); + } + throw new TaskQueueException('Task creation failed: ' . $msg); + } + } + } + $resObj = $response->getResponse(); + if ($resObj) { + foreach ($resObj->getTasks() as $resTask) { + $parts = explode('/', $resTask->getName()); + $names[] = end($parts); + } } } catch (\Google\ApiCore\ApiException $e) { if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409) { From 7ee7e06824405ca60d284df3eb708d162ef8b91c Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 3 Aug 2026 07:44:23 +0000 Subject: [PATCH 24/37] Align Cloud Tasks batchCreateTasks error handling in PushQueue.php with legacy TaskQueue precedence --- src/Api/TaskQueue/PushQueue.php | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 7ec67712..5e9bde37 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -379,7 +379,17 @@ private function addTasksCloudTasks($tasks) { $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); try { $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + + $resObj = $response->getResponse(); + if ($resObj) { + foreach ($resObj->getTasks() as $resTask) { + $parts = explode('/', $resTask->getName()); + $names[] = end($parts); + } + } + $metadata = $response->getMetadata(); + $exception = null; if ($metadata && $metadata->getFailedRequests()) { foreach ($chunk as $idx => $task) { if ($metadata->getFailedRequests()->offsetExists($idx)) { @@ -387,18 +397,15 @@ private function addTasksCloudTasks($tasks) { $code = $errStatus->getCode(); $msg = $errStatus->getMessage(); if (self::isAlreadyExistsError($code, $msg)) { - throw new TaskAlreadyExistsException('Task exists already: ' . $msg); + $exception = new TaskAlreadyExistsException('Task exists already: ' . $msg); + } else { + throw new TaskQueueException('Task creation failed: ' . $msg); } - throw new TaskQueueException('Task creation failed: ' . $msg); } } } - $resObj = $response->getResponse(); - if ($resObj) { - foreach ($resObj->getTasks() as $resTask) { - $parts = explode('/', $resTask->getName()); - $names[] = end($parts); - } + if ($exception !== null) { + throw $exception; } } catch (\Google\ApiCore\ApiException $e) { if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409) { From 34495424cbc6c72222c479e97132487d29d4be40 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 3 Aug 2026 08:06:11 +0000 Subject: [PATCH 25/37] Update google/cloud-tasks dependency to ^2.0 in composer.json for batchCreateTasks support --- composer.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/composer.json b/composer.json index bb3d7ae3..a9e74b5e 100644 --- a/composer.json +++ b/composer.json @@ -14,7 +14,7 @@ "guzzlehttp/streams": "^3.0", "guzzlehttp/guzzle": "^7.2", "composer/semver": "^3.2", - "google/cloud-tasks": "^1.0" + "google/cloud-tasks": "^2.0" }, "require-dev": { "phpunit/phpunit": "^8", From 514a3bd5ff833f18af7035d71a9ff87d2d779ea3 Mon Sep 17 00:00:00 2001 From: Chirag Gajjar Date: Mon, 3 Aug 2026 08:11:36 +0000 Subject: [PATCH 26/37] Use integer enum constants for HttpMethod in PushQueue.php --- src/Api/TaskQueue/PushQueue.php | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 5e9bde37..8cafcadf 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -330,13 +330,15 @@ private function addTasksCloudTasks($tasks) { } $methodMap = [ - 'POST' => \Google\Cloud\Tasks\V2beta3\HttpMethod::POST, - 'GET' => \Google\Cloud\Tasks\V2beta3\HttpMethod::GET, - 'PUT' => \Google\Cloud\Tasks\V2beta3\HttpMethod::PUT, - 'DELETE' => \Google\Cloud\Tasks\V2beta3\HttpMethod::DELETE, - 'HEAD' => \Google\Cloud\Tasks\V2beta3\HttpMethod::HEAD, + 'POST' => 1, + 'GET' => 2, + 'HEAD' => 3, + 'PUT' => 4, + 'DELETE' => 5, + 'PATCH' => 6, + 'OPTIONS' => 7, ]; - $httpMethod = isset($methodMap[$task->getMethod()]) ? $methodMap[$task->getMethod()] : \Google\Cloud\Tasks\V2beta3\HttpMethod::POST; + $httpMethod = isset($methodMap[$task->getMethod()]) ? $methodMap[$task->getMethod()] : 1; $appEngineReq = new \Google\Cloud\Tasks\V2beta3\AppEngineHttpRequest(); $appEngineReq->setRelativeUri($task->getUrl() ?: '/'); From cc200192f1a48884c889c9e6b8b5931e66964085 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Thu, 3 Sep 2026 12:38:54 +0000 Subject: [PATCH 27/37] feat(taskqueue): migrate push queue client to Cloud Tasks v2 --- src/Api/TaskQueue/PushQueue.php | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 8cafcadf..4217cac6 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -340,7 +340,7 @@ private function addTasksCloudTasks($tasks) { ]; $httpMethod = isset($methodMap[$task->getMethod()]) ? $methodMap[$task->getMethod()] : 1; - $appEngineReq = new \Google\Cloud\Tasks\V2beta3\AppEngineHttpRequest(); + $appEngineReq = new \Google\Cloud\Tasks\V2\AppEngineHttpRequest(); $appEngineReq->setRelativeUri($task->getUrl() ?: '/'); $appEngineReq->setHttpMethod($httpMethod); @@ -359,7 +359,7 @@ private function addTasksCloudTasks($tasks) { } } - $taskObj = new \Google\Cloud\Tasks\V2beta3\Task(); + $taskObj = new \Google\Cloud\Tasks\V2\Task(); if ($taskName) { $fullTaskName = $fullQueueName . "/tasks/" . $taskName; $taskObj->setName($fullTaskName); @@ -372,13 +372,13 @@ private function addTasksCloudTasks($tasks) { $taskObj->setScheduleTime($ts); } - $createTaskReq = new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest(); + $createTaskReq = new \Google\Cloud\Tasks\V2\CreateTaskRequest(); $createTaskReq->setParent($fullQueueName); $createTaskReq->setTask($taskObj); $createTaskRequests[] = $createTaskReq; } - $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); + $client = new \Google\Cloud\Tasks\V2\CloudTasksClient(); try { $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); From 600f7416aea762fd957881b7271a02c7febdd062 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Mon, 7 Sep 2026 07:59:14 +0000 Subject: [PATCH 28/37] feat(taskqueue): support both GAE_PUSHQUEUE_BACKEND and APPENGINE_USE_CLOUDTASK_PUSH_QUEUE env vars --- src/Api/TaskQueue/PushQueue.php | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 4217cac6..f1797372 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -148,7 +148,10 @@ public function addTasks($tasks) { ' tasks. Actual size: ' . count($tasks)); } - if (getenv('GAE_PUSHQUEUE_BACKEND') === 'CLOUD_TASK') { + $useCloudTasks = getenv('GAE_PUSHQUEUE_BACKEND') === 'CLOUD_TASK' || + strtolower((string) getenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE')) === 'true' || + getenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE') === '1'; + if ($useCloudTasks) { return $this->addTasksCloudTasks($tasks); } From 77c46501e82f9267e44c39885b76616c89224877 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Mon, 7 Sep 2026 08:11:18 +0000 Subject: [PATCH 29/37] chore(taskqueue): remove unused token lookup and revert AppIdentityService --- src/Api/AppIdentity/AppIdentityService.php | 37 +++++++--------------- src/Api/TaskQueue/PushQueue.php | 20 ------------ 2 files changed, 11 insertions(+), 46 deletions(-) diff --git a/src/Api/AppIdentity/AppIdentityService.php b/src/Api/AppIdentity/AppIdentityService.php index 747bd73c..077d3bb2 100644 --- a/src/Api/AppIdentity/AppIdentityService.php +++ b/src/Api/AppIdentity/AppIdentityService.php @@ -274,12 +274,8 @@ public static function getDefaultVersionHostname() { private static function putTokenInCache($name, $value, $expiry_secs) { $expiry_time_from_epoch = $expiry_secs - self::EXPIRY_SAFETY_MARGIN_SECS - self::EXPIRY_SHORT_MARGIN_SECS; - if (class_exists('Memcache')) { - try { - $memcache = new Memcache(); - $memcache->set($name, $value, null, $expiry_time_from_epoch); - } catch (\Throwable $t) {} - } + $memcache = new Memcache(); + $memcache->set($name, $value, null, $expiry_time_from_epoch); // Record the expiry time in the object being cached, so we can check it // when read from APC. self::putTokenInApc($name, $value, $expiry_secs); @@ -300,11 +296,7 @@ private static function putTokenInApc($name, $value, $expiry_secs) { self::EXPIRY_SHORT_MARGIN_SECS; $cache_ttl = self::getTTLForToken($expiry_time_from_epoch); $value['eviction_time_epoch'] = $cache_ttl['eviction_time_epoch']; - if (function_exists('apcu_store')) { - apcu_store($name, $value, $cache_ttl['apc_ttl_in_seconds']); - } elseif (function_exists('apc_store')) { - apc_store($name, $value, $cache_ttl['apc_ttl_in_seconds']); - } + apc_store($name, $value, $cache_ttl['apc_ttl_in_seconds']); } /** @@ -319,24 +311,17 @@ private static function putTokenInApc($name, $value, $expiry_secs) { */ private static function getTokenFromCache($name) { $success = false; - $result = false; - if (function_exists('apcu_fetch')) { - $result = apcu_fetch($name, $success); - } elseif (function_exists('apc_fetch')) { - $result = apc_fetch($name, $success); - } - if ($success && $result !== false && time() < $result['eviction_time_epoch']) { + $result = apc_fetch($name, $success); + if ($success && time() < $result['eviction_time_epoch']) { unset($result['eviction_time_epoch']); return $result; } - if (class_exists('Memcache')) { - try { - $memcache = new Memcache(); - $result = $memcache->get($name); - if ($result !== false) { - self::putTokenInApc($name, $result, $result['expiration_time']); - } - } catch (\Throwable $t) {} + $memcache = new Memcache(); + $result = $memcache->get($name); + // If there was a result in memcache but not in apc we can add using a + // short timeout. + if ($result !== false) { + self::putTokenInApc($name, $result, $result['expiration_time']); } return $result; } diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index f1797372..1d684d75 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -272,29 +272,9 @@ private static function getProjectId() { return $projectId; } - private static function getCloudPlatformToken() { - try { - $res = \Google\AppEngine\Api\AppIdentity\AppIdentityService::getAccessToken('https://www.googleapis.com/auth/cloud-platform'); - if (isset($res['access_token'])) { - return $res['access_token']; - } - } catch (\Exception $e) { - // Fallback to metadata server if AppIdentityService fails - } - $json = self::getMetadataValue('instance/service-accounts/default/token'); - if ($json) { - $data = json_decode($json, true); - if (isset($data['access_token'])) { - return $data['access_token']; - } - } - throw new TaskQueueException('Failed to obtain OAuth access token for Cloud Tasks'); - } - private function addTasksCloudTasks($tasks) { $projectId = self::getProjectId(); $region = self::getRegion(); - $token = self::getCloudPlatformToken(); $fullQueueName = "projects/" . $projectId . "/locations/" . $region . "/queues/" . $this->name; $names = []; From 50d1798b3329e1c585b9f65cb9b0974e513d30b7 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Mon, 7 Sep 2026 08:24:17 +0000 Subject: [PATCH 30/37] feat(taskqueue): route single tasks to createTask and batches to batchCreateTasks --- src/Api/TaskQueue/PushQueue.php | 168 ++++++++++++++++++-------------- 1 file changed, 97 insertions(+), 71 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 1d684d75..e1dc9714 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -272,11 +272,107 @@ private static function getProjectId() { return $projectId; } + private function buildCloudTaskObj($task, $fullQueueName) { + $headers = []; + $hasContentType = false; + foreach ($task->getHeaders() as $header) { + $pair = explode(':', $header, 2); + $key = trim($pair[0]); + $val = trim($pair[1]); + if (strcasecmp($key, 'Content-Type') === 0) { + $hasContentType = true; + $key = 'Content-Type'; + } elseif (strcasecmp($key, 'X-AppEngine-QueueName') === 0) { + $key = 'X-AppEngine-QueueName'; + } elseif (strcasecmp($key, 'X-AppEngine-TaskName') === 0) { + $key = 'X-AppEngine-TaskName'; + } + $headers[$key] = $val; + } + if (!$hasContentType) { + $headers['Content-Type'] = 'application/octet-stream'; + } + if (!isset($headers['X-AppEngine-QueueName'])) { + $headers['X-AppEngine-QueueName'] = $this->name; + } + + $taskName = $task->getName(); + if ($taskName && !isset($headers['X-AppEngine-TaskName'])) { + $headers['X-AppEngine-TaskName'] = $taskName; + } + + $methodMap = [ + 'POST' => 1, + 'GET' => 2, + 'HEAD' => 3, + 'PUT' => 4, + 'DELETE' => 5, + 'PATCH' => 6, + 'OPTIONS' => 7, + ]; + $httpMethod = isset($methodMap[$task->getMethod()]) ? $methodMap[$task->getMethod()] : 1; + + $appEngineReq = new \Google\Cloud\Tasks\V2\AppEngineHttpRequest(); + $appEngineReq->setRelativeUri($task->getUrl() ?: '/'); + $appEngineReq->setHttpMethod($httpMethod); + + foreach ($headers as $k => $v) { + $appEngineReq->getHeaders()[$k] = $v; + } + + if ($task->getMethod() === 'POST' || $task->getMethod() === 'PUT') { + if ($task->getQueryData()) { + $body = http_build_query($task->getQueryData()); + if (strlen($body) > PushTask::MAX_TASK_SIZE_BYTES) { + throw new TaskQueueException('Task greater than maximum size of ' . + PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); + } + $appEngineReq->setBody($body); + } + } + + $taskObj = new \Google\Cloud\Tasks\V2\Task(); + if ($taskName) { + $fullTaskName = $fullQueueName . "/tasks/" . $taskName; + $taskObj->setName($fullTaskName); + } + $taskObj->setAppEngineHttpRequest($appEngineReq); + + if ($task->getDelaySeconds() > 0) { + $ts = new \Google\Protobuf\Timestamp(); + $ts->setSeconds(time() + $task->getDelaySeconds()); + $taskObj->setScheduleTime($ts); + } + + return $taskObj; + } + + private function createSingleTaskCloudTasks($task, $fullQueueName) { + $taskObj = $this->buildCloudTaskObj($task, $fullQueueName); + $client = new \Google\Cloud\Tasks\V2\CloudTasksClient(); + try { + $response = $client->createTask($fullQueueName, $taskObj); + $parts = explode('/', $response->getName()); + return [end($parts)]; + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); + } finally { + $client->close(); + } + } + private function addTasksCloudTasks($tasks) { $projectId = self::getProjectId(); $region = self::getRegion(); $fullQueueName = "projects/" . $projectId . "/locations/" . $region . "/queues/" . $this->name; + if (count($tasks) === 1) { + return $this->createSingleTaskCloudTasks($tasks[0], $fullQueueName); + } + $names = []; $chunks = array_chunk($tasks, 100); @@ -284,77 +380,7 @@ private function addTasksCloudTasks($tasks) { $createTaskRequests = []; foreach ($chunk as $task) { - $headers = []; - $hasContentType = false; - foreach ($task->getHeaders() as $header) { - $pair = explode(':', $header, 2); - $key = trim($pair[0]); - $val = trim($pair[1]); - if (strcasecmp($key, 'Content-Type') === 0) { - $hasContentType = true; - $key = 'Content-Type'; - } elseif (strcasecmp($key, 'X-AppEngine-QueueName') === 0) { - $key = 'X-AppEngine-QueueName'; - } elseif (strcasecmp($key, 'X-AppEngine-TaskName') === 0) { - $key = 'X-AppEngine-TaskName'; - } - $headers[$key] = $val; - } - if (!$hasContentType) { - $headers['Content-Type'] = 'application/octet-stream'; - } - if (!isset($headers['X-AppEngine-QueueName'])) { - $headers['X-AppEngine-QueueName'] = $this->name; - } - - $taskName = $task->getName(); - if ($taskName && !isset($headers['X-AppEngine-TaskName'])) { - $headers['X-AppEngine-TaskName'] = $taskName; - } - - $methodMap = [ - 'POST' => 1, - 'GET' => 2, - 'HEAD' => 3, - 'PUT' => 4, - 'DELETE' => 5, - 'PATCH' => 6, - 'OPTIONS' => 7, - ]; - $httpMethod = isset($methodMap[$task->getMethod()]) ? $methodMap[$task->getMethod()] : 1; - - $appEngineReq = new \Google\Cloud\Tasks\V2\AppEngineHttpRequest(); - $appEngineReq->setRelativeUri($task->getUrl() ?: '/'); - $appEngineReq->setHttpMethod($httpMethod); - - foreach ($headers as $k => $v) { - $appEngineReq->getHeaders()[$k] = $v; - } - - if ($task->getMethod() === 'POST' || $task->getMethod() === 'PUT') { - if ($task->getQueryData()) { - $body = http_build_query($task->getQueryData()); - if (strlen($body) > PushTask::MAX_TASK_SIZE_BYTES) { - throw new TaskQueueException('Task greater than maximum size of ' . - PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); - } - $appEngineReq->setBody($body); - } - } - - $taskObj = new \Google\Cloud\Tasks\V2\Task(); - if ($taskName) { - $fullTaskName = $fullQueueName . "/tasks/" . $taskName; - $taskObj->setName($fullTaskName); - } - $taskObj->setAppEngineHttpRequest($appEngineReq); - - if ($task->getDelaySeconds() > 0) { - $ts = new \Google\Protobuf\Timestamp(); - $ts->setSeconds(time() + $task->getDelaySeconds()); - $taskObj->setScheduleTime($ts); - } - + $taskObj = $this->buildCloudTaskObj($task, $fullQueueName); $createTaskReq = new \Google\Cloud\Tasks\V2\CreateTaskRequest(); $createTaskReq->setParent($fullQueueName); $createTaskReq->setTask($taskObj); From 6a5349a032b98e5cc9fdf3dabf2dae075b761b11 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Mon, 7 Sep 2026 17:02:25 +0000 Subject: [PATCH 31/37] Support CloudTasksClient V2 client namespace and V2beta3 batchCreateTasks fallback --- src/Api/TaskQueue/PushQueue.php | 239 ++++++++++++++++++++++++-------- 1 file changed, 185 insertions(+), 54 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index e1dc9714..460e864b 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -272,7 +272,7 @@ private static function getProjectId() { return $projectId; } - private function buildCloudTaskObj($task, $fullQueueName) { + private function buildCloudTaskObjV2($task, $fullQueueName) { $headers = []; $hasContentType = false; foreach ($task->getHeaders() as $header) { @@ -347,21 +347,117 @@ private function buildCloudTaskObj($task, $fullQueueName) { return $taskObj; } + private function buildCloudTaskObjV2beta3($task, $fullQueueName) { + $headers = []; + $hasContentType = false; + foreach ($task->getHeaders() as $header) { + $pair = explode(':', $header, 2); + $key = trim($pair[0]); + $val = trim($pair[1]); + if (strcasecmp($key, 'Content-Type') === 0) { + $hasContentType = true; + $key = 'Content-Type'; + } elseif (strcasecmp($key, 'X-AppEngine-QueueName') === 0) { + $key = 'X-AppEngine-QueueName'; + } elseif (strcasecmp($key, 'X-AppEngine-TaskName') === 0) { + $key = 'X-AppEngine-TaskName'; + } + $headers[$key] = $val; + } + if (!$hasContentType) { + $headers['Content-Type'] = 'application/octet-stream'; + } + if (!isset($headers['X-AppEngine-QueueName'])) { + $headers['X-AppEngine-QueueName'] = $this->name; + } + + $taskName = $task->getName(); + if ($taskName && !isset($headers['X-AppEngine-TaskName'])) { + $headers['X-AppEngine-TaskName'] = $taskName; + } + + $methodMap = [ + 'POST' => 1, + 'GET' => 2, + 'HEAD' => 3, + 'PUT' => 4, + 'DELETE' => 5, + 'PATCH' => 6, + 'OPTIONS' => 7, + ]; + $httpMethod = isset($methodMap[$task->getMethod()]) ? $methodMap[$task->getMethod()] : 1; + + $appEngineReq = new \Google\Cloud\Tasks\V2beta3\AppEngineHttpRequest(); + $appEngineReq->setRelativeUri($task->getUrl() ?: '/'); + $appEngineReq->setHttpMethod($httpMethod); + + foreach ($headers as $k => $v) { + $appEngineReq->getHeaders()[$k] = $v; + } + + if ($task->getMethod() === 'POST' || $task->getMethod() === 'PUT') { + if ($task->getQueryData()) { + $body = http_build_query($task->getQueryData()); + if (strlen($body) > PushTask::MAX_TASK_SIZE_BYTES) { + throw new TaskQueueException('Task greater than maximum size of ' . + PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); + } + $appEngineReq->setBody($body); + } + } + + $taskObj = new \Google\Cloud\Tasks\V2beta3\Task(); + if ($taskName) { + $fullTaskName = $fullQueueName . "/tasks/" . $taskName; + $taskObj->setName($fullTaskName); + } + $taskObj->setAppEngineHttpRequest($appEngineReq); + + if ($task->getDelaySeconds() > 0) { + $ts = new \Google\Protobuf\Timestamp(); + $ts->setSeconds(time() + $task->getDelaySeconds()); + $taskObj->setScheduleTime($ts); + } + + return $taskObj; + } + private function createSingleTaskCloudTasks($task, $fullQueueName) { - $taskObj = $this->buildCloudTaskObj($task, $fullQueueName); - $client = new \Google\Cloud\Tasks\V2\CloudTasksClient(); - try { - $response = $client->createTask($fullQueueName, $taskObj); - $parts = explode('/', $response->getName()); - return [end($parts)]; - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + if (class_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient')) { + $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); + $client = new \Google\Cloud\Tasks\V2\Client\CloudTasksClient(); + $createTaskReq = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) + ->setParent($fullQueueName) + ->setTask($taskObj); + try { + $response = $client->createTask($createTaskReq); + $parts = explode('/', $response->getName()); + return [end($parts)]; + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); + } finally { + $client->close(); + } + } elseif (class_exists('\Google\Cloud\Tasks\V2beta3\CloudTasksClient')) { + $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); + $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); + try { + $response = $client->createTask($fullQueueName, $taskObj); + $parts = explode('/', $response->getName()); + return [end($parts)]; + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); + } finally { + $client->close(); } - throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); - } finally { - $client->close(); } + throw new TaskQueueException('Cloud Tasks Client SDK is not available.'); } private function addTasksCloudTasks($tasks) { @@ -376,66 +472,101 @@ private function addTasksCloudTasks($tasks) { $names = []; $chunks = array_chunk($tasks, 100); - foreach ($chunks as $chunk) { - $createTaskRequests = []; + if (class_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient') && method_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient', 'batchCreateTasks')) { + foreach ($chunks as $chunk) { + $createTaskRequests = []; + foreach ($chunk as $task) { + $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); + $createTaskReq = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) + ->setParent($fullQueueName) + ->setTask($taskObj); + $createTaskRequests[] = $createTaskReq; + } - foreach ($chunk as $task) { - $taskObj = $this->buildCloudTaskObj($task, $fullQueueName); - $createTaskReq = new \Google\Cloud\Tasks\V2\CreateTaskRequest(); - $createTaskReq->setParent($fullQueueName); - $createTaskReq->setTask($taskObj); - $createTaskRequests[] = $createTaskReq; + $client = new \Google\Cloud\Tasks\V2\Client\CloudTasksClient(); + try { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + $resObj = method_exists($response, 'getResponse') ? $response->getResponse() : $response; + if ($resObj && method_exists($resObj, 'getTasks')) { + foreach ($resObj->getTasks() as $resTask) { + $parts = explode('/', $resTask->getName()); + $names[] = end($parts); + } + } + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK batchCreate failed: ' . $e->getMessage()); + } finally { + $client->close(); + } } + return $names; + } - $client = new \Google\Cloud\Tasks\V2\CloudTasksClient(); - try { - $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + if (class_exists('\Google\Cloud\Tasks\V2beta3\CloudTasksClient')) { + foreach ($chunks as $chunk) { + $createTaskRequests = []; + foreach ($chunk as $task) { + $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); + $createTaskReq = new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest(); + $createTaskReq->setParent($fullQueueName); + $createTaskReq->setTask($taskObj); + $createTaskRequests[] = $createTaskReq; + } - $resObj = $response->getResponse(); - if ($resObj) { - foreach ($resObj->getTasks() as $resTask) { - $parts = explode('/', $resTask->getName()); - $names[] = end($parts); + $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); + try { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + + $resObj = $response->getResponse(); + if ($resObj) { + foreach ($resObj->getTasks() as $resTask) { + $parts = explode('/', $resTask->getName()); + $names[] = end($parts); + } } - } - $metadata = $response->getMetadata(); - $exception = null; - if ($metadata && $metadata->getFailedRequests()) { - foreach ($chunk as $idx => $task) { - if ($metadata->getFailedRequests()->offsetExists($idx)) { - $errStatus = $metadata->getFailedRequests()->offsetGet($idx); - $code = $errStatus->getCode(); - $msg = $errStatus->getMessage(); - if (self::isAlreadyExistsError($code, $msg)) { - $exception = new TaskAlreadyExistsException('Task exists already: ' . $msg); - } else { - throw new TaskQueueException('Task creation failed: ' . $msg); + $metadata = $response->getMetadata(); + $exception = null; + if ($metadata && $metadata->getFailedRequests()) { + foreach ($chunk as $idx => $task) { + if ($metadata->getFailedRequests()->offsetExists($idx)) { + $errStatus = $metadata->getFailedRequests()->offsetGet($idx); + $code = $errStatus->getCode(); + $msg = $errStatus->getMessage(); + if (self::isAlreadyExistsError($code, $msg)) { + $exception = new TaskAlreadyExistsException('Task exists already: ' . $msg); + } else { + throw new TaskQueueException('Task creation failed: ' . $msg); + } } } } + if ($exception !== null) { + throw $exception; + } + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK batchCreate failed: ' . $e->getMessage()); + } finally { + $client->close(); } - if ($exception !== null) { - throw $exception; - } - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); - } - throw new TaskQueueException('Cloud Tasks Client SDK batchCreate failed: ' . $e->getMessage()); - } finally { - $client->close(); } + return $names; } - return $names; + throw new TaskQueueException('Cloud Tasks Client SDK batchCreate is not available.'); } private static function isAlreadyExistsError($errCode, $errMsg) { if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false) { return true; } - if (($errCode === 5 || $errCode === 404) && stripos($errMsg, 'Requested entity was not found') !== false) { + if (($errCode === 5 || $errCode === 404) && (stripos($errMsg, 'Requested entity was not found') !== false || stripos($errMsg, 'tombstoned') !== false)) { return true; } return false; From 56ba2d0ae32ceae7b02446787f397a7fc565a07c Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Mon, 7 Sep 2026 17:06:19 +0000 Subject: [PATCH 32/37] Support V2beta3 Client namespace and BatchCreateTasksRequest in addTasksCloudTasks --- src/Api/TaskQueue/PushQueue.php | 96 +++++++++++++++++++++++++++++---- 1 file changed, 87 insertions(+), 9 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 460e864b..d5bc7263 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -441,6 +441,39 @@ private function createSingleTaskCloudTasks($task, $fullQueueName) { } finally { $client->close(); } + } elseif (class_exists('\Google\Cloud\Tasks\V2\CloudTasksClient')) { + $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); + $client = new \Google\Cloud\Tasks\V2\CloudTasksClient(); + try { + $response = $client->createTask($fullQueueName, $taskObj); + $parts = explode('/', $response->getName()); + return [end($parts)]; + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); + } finally { + $client->close(); + } + } elseif (class_exists('\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient')) { + $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); + $client = new \Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient(); + $createTaskReq = (new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest()) + ->setParent($fullQueueName) + ->setTask($taskObj); + try { + $response = $client->createTask($createTaskReq); + $parts = explode('/', $response->getName()); + return [end($parts)]; + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); + } finally { + $client->close(); + } } elseif (class_exists('\Google\Cloud\Tasks\V2beta3\CloudTasksClient')) { $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); @@ -472,7 +505,15 @@ private function addTasksCloudTasks($tasks) { $names = []; $chunks = array_chunk($tasks, 100); + // 1. Try V2 batchCreateTasks if available + $v2ClientClass = null; if (class_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient') && method_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient', 'batchCreateTasks')) { + $v2ClientClass = '\Google\Cloud\Tasks\V2\Client\CloudTasksClient'; + } elseif (class_exists('\Google\Cloud\Tasks\V2\CloudTasksClient') && method_exists('\Google\Cloud\Tasks\V2\CloudTasksClient', 'batchCreateTasks')) { + $v2ClientClass = '\Google\Cloud\Tasks\V2\CloudTasksClient'; + } + + if ($v2ClientClass !== null) { foreach ($chunks as $chunk) { $createTaskRequests = []; foreach ($chunk as $task) { @@ -483,9 +524,24 @@ private function addTasksCloudTasks($tasks) { $createTaskRequests[] = $createTaskReq; } - $client = new \Google\Cloud\Tasks\V2\Client\CloudTasksClient(); + $client = new $v2ClientClass(); try { - $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + if (class_exists('\Google\Cloud\Tasks\V2\BatchCreateTasksRequest')) { + $batchReq = (new \Google\Cloud\Tasks\V2\BatchCreateTasksRequest()) + ->setParent($fullQueueName) + ->setRequests($createTaskRequests); + try { + $response = $client->batchCreateTasks($batchReq); + } catch (\TypeError $te) { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + } + } else { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + } + + if (method_exists($response, 'pollUntilComplete') && !$response->isDone()) { + $response->pollUntilComplete(); + } $resObj = method_exists($response, 'getResponse') ? $response->getResponse() : $response; if ($resObj && method_exists($resObj, 'getTasks')) { foreach ($resObj->getTasks() as $resTask) { @@ -505,7 +561,15 @@ private function addTasksCloudTasks($tasks) { return $names; } - if (class_exists('\Google\Cloud\Tasks\V2beta3\CloudTasksClient')) { + // 2. Try V2beta3 batchCreateTasks + $betaClientClass = null; + if (class_exists('\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient') && method_exists('\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient', 'batchCreateTasks')) { + $betaClientClass = '\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient'; + } elseif (class_exists('\Google\Cloud\Tasks\V2beta3\CloudTasksClient') && method_exists('\Google\Cloud\Tasks\V2beta3\CloudTasksClient', 'batchCreateTasks')) { + $betaClientClass = '\Google\Cloud\Tasks\V2beta3\CloudTasksClient'; + } + + if ($betaClientClass !== null) { foreach ($chunks as $chunk) { $createTaskRequests = []; foreach ($chunk as $task) { @@ -516,21 +580,35 @@ private function addTasksCloudTasks($tasks) { $createTaskRequests[] = $createTaskReq; } - $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); + $client = new $betaClientClass(); try { - $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + if (class_exists('\Google\Cloud\Tasks\V2beta3\BatchCreateTasksRequest')) { + $batchReq = (new \Google\Cloud\Tasks\V2beta3\BatchCreateTasksRequest()) + ->setParent($fullQueueName) + ->setRequests($createTaskRequests); + try { + $response = $client->batchCreateTasks($batchReq); + } catch (\TypeError $te) { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + } + } else { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + } - $resObj = $response->getResponse(); - if ($resObj) { + if (method_exists($response, 'pollUntilComplete') && !$response->isDone()) { + $response->pollUntilComplete(); + } + $resObj = method_exists($response, 'getResponse') ? $response->getResponse() : $response; + if ($resObj && method_exists($resObj, 'getTasks')) { foreach ($resObj->getTasks() as $resTask) { $parts = explode('/', $resTask->getName()); $names[] = end($parts); } } - $metadata = $response->getMetadata(); + $metadata = method_exists($response, 'getMetadata') ? $response->getMetadata() : null; $exception = null; - if ($metadata && $metadata->getFailedRequests()) { + if ($metadata && method_exists($metadata, 'getFailedRequests') && $metadata->getFailedRequests()) { foreach ($chunk as $idx => $task) { if ($metadata->getFailedRequests()->offsetExists($idx)) { $errStatus = $metadata->getFailedRequests()->offsetGet($idx); From 456d14939346925e26004bcdb26b10993d969777 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Mon, 7 Sep 2026 17:15:09 +0000 Subject: [PATCH 33/37] Fallback to createTask for batch tasks when batchCreateTasks is not available in SDK --- src/Api/TaskQueue/PushQueue.php | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index d5bc7263..2a925724 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -637,7 +637,13 @@ private function addTasksCloudTasks($tasks) { return $names; } - throw new TaskQueueException('Cloud Tasks Client SDK batchCreate is not available.'); + // Fallback: If native batchCreateTasks is not available in the installed Cloud Tasks SDK, + // enqueue tasks individually using createTask. + foreach ($tasks as $task) { + $res = $this->createSingleTaskCloudTasks($task, $fullQueueName); + $names[] = $res[0]; + } + return $names; } private static function isAlreadyExistsError($errCode, $errMsg) { From 5dae248728d9e43e4984fa395b2c3dc3a5fb2764 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Wed, 9 Sep 2026 06:32:27 +0000 Subject: [PATCH 34/37] fix(composer): move google/cloud-tasks to suggest and pin php-coveralls to ^2.7 for PHP 7.2-8.2 compatibility --- composer.json | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/composer.json b/composer.json index a9e74b5e..ce08150a 100644 --- a/composer.json +++ b/composer.json @@ -13,14 +13,16 @@ "php": ">=7.2.0", "guzzlehttp/streams": "^3.0", "guzzlehttp/guzzle": "^7.2", - "composer/semver": "^3.2", - "google/cloud-tasks": "^2.0" + "composer/semver": "^3.2" }, "require-dev": { "phpunit/phpunit": "^8", - "php-coveralls/php-coveralls": "dev-master", + "php-coveralls/php-coveralls": "^2.7", "php-mock/php-mock-phpunit": "^2.6" }, + "suggest": { + "google/cloud-tasks": "Required to use Cloud Tasks push queue routing (^2.0)." + }, "autoload": { "psr-4": { "Google\\AppEngine\\": "src" From b2a9a2bf139f4d9e271fd64a8cad3a12d89db638 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Tue, 6 Oct 2026 05:32:25 +0000 Subject: [PATCH 35/37] Support OperationResponse::getResult and failedRequests metadata for Cloud Tasks v2 batchCreateTasks --- composer.json | 2 +- src/Api/TaskQueue/PushQueue.php | 157 +++++++++++++++----------- tests/Api/TaskQueue/PushQueueTest.php | 86 ++++++++++++++ 3 files changed, 181 insertions(+), 64 deletions(-) diff --git a/composer.json b/composer.json index ce08150a..6cc51e89 100644 --- a/composer.json +++ b/composer.json @@ -21,7 +21,7 @@ "php-mock/php-mock-phpunit": "^2.6" }, "suggest": { - "google/cloud-tasks": "Required to use Cloud Tasks push queue routing (^2.0)." + "google/cloud-tasks": "Required to use Cloud Tasks push queue routing (^2.3)." }, "autoload": { "psr-4": { diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 2a925724..0248fa6e 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -50,6 +50,19 @@ final class PushQueue { 'DELETE' => RequestMethod::DELETE ]; + private static $cloudTasksClientFactory = null; + + public static function setCloudTasksClientFactory($factory) { + self::$cloudTasksClientFactory = $factory; + } + + private static function newCloudTasksClient($clientClass) { + if (self::$cloudTasksClientFactory !== null) { + return call_user_func(self::$cloudTasksClientFactory, $clientClass); + } + return new $clientClass(); + } + /** * Construct a PushQueue * @@ -147,6 +160,13 @@ public function addTasks($tasks) { '$tasks must contain at most ' . self::MAX_TASKS_PER_ADD . ' tasks. Actual size: ' . count($tasks)); } + foreach ($tasks as $task) { + if (!($task instanceof PushTask)) { + throw new \InvalidArgumentException( + 'All values in $tasks must be instances of PushTask. ' . + 'Actual type: ' . gettype($task)); + } + } $useCloudTasks = getenv('GAE_PUSHQUEUE_BACKEND') === 'CLOUD_TASK' || strtolower((string) getenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE')) === 'true' || @@ -161,11 +181,6 @@ public function addTasks($tasks) { $names = []; $current_time = microtime(true); foreach ($tasks as $task) { - if (!($task instanceof PushTask)) { - throw new \InvalidArgumentException( - 'All values in $tasks must be instances of PushTask. ' . - 'Actual type: ' . gettype($task)); - } $names[] = $task->getName(); $add = $req->addAddRequest(); $add->setQueueName($this->name); @@ -236,15 +251,16 @@ private static function getMetadataValue($path) { } private static function getRegion() { + $envRegion = getenv('LOCATION_ID') ?: getenv('GAE_LOCATION') ?: getenv('GAE_REGION') ?: getenv('REGION_ID'); + if ($envRegion) { + return $envRegion; + } static $region = null; if ($region === null) { - $region = getenv('LOCATION_ID') ?: getenv('GAE_LOCATION') ?: getenv('GAE_REGION') ?: getenv('REGION_ID'); - if (!$region) { - $regionPath = self::getMetadataValue('instance/region'); - if ($regionPath) { - $parts = explode('/', $regionPath); - $region = end($parts); - } + $regionPath = self::getMetadataValue('instance/region'); + if ($regionPath) { + $parts = explode('/', $regionPath); + $region = end($parts); } if (!$region) { $region = getenv('LOCAL_GCP_REGION') ?: 'us-central1'; @@ -254,12 +270,13 @@ private static function getRegion() { } private static function getProjectId() { + $envProject = getenv('GOOGLE_CLOUD_PROJECT'); + if ($envProject) { + return $envProject; + } static $projectId = null; if ($projectId === null) { - $projectId = getenv('GOOGLE_CLOUD_PROJECT'); - if (!$projectId) { - $projectId = self::getMetadataValue('project/project-id'); - } + $projectId = self::getMetadataValue('project/project-id'); if (!$projectId) { $appId = ApiProxy::getCurrentAppId(); if (($pos = strpos($appId, '~')) !== false) { @@ -425,7 +442,7 @@ private function buildCloudTaskObjV2beta3($task, $fullQueueName) { private function createSingleTaskCloudTasks($task, $fullQueueName) { if (class_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient')) { $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); - $client = new \Google\Cloud\Tasks\V2\Client\CloudTasksClient(); + $client = self::newCloudTasksClient('\Google\Cloud\Tasks\V2\Client\CloudTasksClient'); $createTaskReq = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) ->setParent($fullQueueName) ->setTask($taskObj); @@ -443,7 +460,7 @@ private function createSingleTaskCloudTasks($task, $fullQueueName) { } } elseif (class_exists('\Google\Cloud\Tasks\V2\CloudTasksClient')) { $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); - $client = new \Google\Cloud\Tasks\V2\CloudTasksClient(); + $client = self::newCloudTasksClient('\Google\Cloud\Tasks\V2\CloudTasksClient'); try { $response = $client->createTask($fullQueueName, $taskObj); $parts = explode('/', $response->getName()); @@ -458,7 +475,7 @@ private function createSingleTaskCloudTasks($task, $fullQueueName) { } } elseif (class_exists('\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient')) { $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); - $client = new \Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient(); + $client = self::newCloudTasksClient('\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient'); $createTaskReq = (new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest()) ->setParent($fullQueueName) ->setTask($taskObj); @@ -476,7 +493,7 @@ private function createSingleTaskCloudTasks($task, $fullQueueName) { } } elseif (class_exists('\Google\Cloud\Tasks\V2beta3\CloudTasksClient')) { $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); - $client = new \Google\Cloud\Tasks\V2beta3\CloudTasksClient(); + $client = self::newCloudTasksClient('\Google\Cloud\Tasks\V2beta3\CloudTasksClient'); try { $response = $client->createTask($fullQueueName, $taskObj); $parts = explode('/', $response->getName()); @@ -493,6 +510,58 @@ private function createSingleTaskCloudTasks($task, $fullQueueName) { throw new TaskQueueException('Cloud Tasks Client SDK is not available.'); } + private static function processBatchCreateResponse($response, $chunk, &$names) { + if (method_exists($response, 'pollUntilComplete') && !$response->isDone()) { + $response->pollUntilComplete(); + } + + $metadata = method_exists($response, 'getMetadata') ? $response->getMetadata() : null; + $failedRequests = ($metadata && method_exists($metadata, 'getFailedRequests')) ? $metadata->getFailedRequests() : null; + $exception = null; + if ($failedRequests) { + foreach ($chunk as $idx => $task) { + $hasFailure = (is_array($failedRequests) && isset($failedRequests[$idx])) || + (is_object($failedRequests) && method_exists($failedRequests, 'offsetExists') && $failedRequests->offsetExists($idx)); + if ($hasFailure) { + $errStatus = is_array($failedRequests) ? $failedRequests[$idx] : $failedRequests->offsetGet($idx); + $code = ($errStatus && method_exists($errStatus, 'getCode')) ? $errStatus->getCode() : 0; + if ($code !== 0) { + $msg = ($errStatus && method_exists($errStatus, 'getMessage')) ? $errStatus->getMessage() : ''; + if (self::isAlreadyExistsError($code, $msg)) { + $exception = new TaskAlreadyExistsException('Task exists already: ' . $msg); + } elseif (stripos($msg, 'queue does not exist') !== false || stripos($msg, 'queue no longer exists') !== false) { + throw new TaskQueueException('Unknown queue: ' . $msg); + } else { + throw new TaskQueueException('Task creation failed: ' . $msg); + } + } + } + } + } + if ($exception !== null) { + throw $exception; + } + + if (method_exists($response, 'operationFailed') && $response->operationFailed()) { + $errStatus = method_exists($response, 'getError') ? $response->getError() : null; + $code = ($errStatus && method_exists($errStatus, 'getCode')) ? $errStatus->getCode() : 0; + $msg = ($errStatus && method_exists($errStatus, 'getMessage')) ? $errStatus->getMessage() : 'Batch operation failed'; + if (self::isAlreadyExistsError($code, $msg)) { + throw new TaskAlreadyExistsException('Task exists already: ' . $msg); + } + throw new TaskQueueException('Task creation failed: ' . $msg); + } + + $resObj = method_exists($response, 'getResult') ? $response->getResult() : + (method_exists($response, 'getResponse') ? $response->getResponse() : $response); + if ($resObj && method_exists($resObj, 'getTasks')) { + foreach ($resObj->getTasks() as $resTask) { + $parts = explode('/', $resTask->getName()); + $names[] = end($parts); + } + } + } + private function addTasksCloudTasks($tasks) { $projectId = self::getProjectId(); $region = self::getRegion(); @@ -524,7 +593,7 @@ private function addTasksCloudTasks($tasks) { $createTaskRequests[] = $createTaskReq; } - $client = new $v2ClientClass(); + $client = self::newCloudTasksClient($v2ClientClass); try { if (class_exists('\Google\Cloud\Tasks\V2\BatchCreateTasksRequest')) { $batchReq = (new \Google\Cloud\Tasks\V2\BatchCreateTasksRequest()) @@ -539,16 +608,7 @@ private function addTasksCloudTasks($tasks) { $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); } - if (method_exists($response, 'pollUntilComplete') && !$response->isDone()) { - $response->pollUntilComplete(); - } - $resObj = method_exists($response, 'getResponse') ? $response->getResponse() : $response; - if ($resObj && method_exists($resObj, 'getTasks')) { - foreach ($resObj->getTasks() as $resTask) { - $parts = explode('/', $resTask->getName()); - $names[] = end($parts); - } - } + self::processBatchCreateResponse($response, $chunk, $names); } catch (\Google\ApiCore\ApiException $e) { if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); @@ -580,7 +640,7 @@ private function addTasksCloudTasks($tasks) { $createTaskRequests[] = $createTaskReq; } - $client = new $betaClientClass(); + $client = self::newCloudTasksClient($betaClientClass); try { if (class_exists('\Google\Cloud\Tasks\V2beta3\BatchCreateTasksRequest')) { $batchReq = (new \Google\Cloud\Tasks\V2beta3\BatchCreateTasksRequest()) @@ -595,36 +655,7 @@ private function addTasksCloudTasks($tasks) { $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); } - if (method_exists($response, 'pollUntilComplete') && !$response->isDone()) { - $response->pollUntilComplete(); - } - $resObj = method_exists($response, 'getResponse') ? $response->getResponse() : $response; - if ($resObj && method_exists($resObj, 'getTasks')) { - foreach ($resObj->getTasks() as $resTask) { - $parts = explode('/', $resTask->getName()); - $names[] = end($parts); - } - } - - $metadata = method_exists($response, 'getMetadata') ? $response->getMetadata() : null; - $exception = null; - if ($metadata && method_exists($metadata, 'getFailedRequests') && $metadata->getFailedRequests()) { - foreach ($chunk as $idx => $task) { - if ($metadata->getFailedRequests()->offsetExists($idx)) { - $errStatus = $metadata->getFailedRequests()->offsetGet($idx); - $code = $errStatus->getCode(); - $msg = $errStatus->getMessage(); - if (self::isAlreadyExistsError($code, $msg)) { - $exception = new TaskAlreadyExistsException('Task exists already: ' . $msg); - } else { - throw new TaskQueueException('Task creation failed: ' . $msg); - } - } - } - } - if ($exception !== null) { - throw $exception; - } + self::processBatchCreateResponse($response, $chunk, $names); } catch (\Google\ApiCore\ApiException $e) { if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); @@ -647,7 +678,7 @@ private function addTasksCloudTasks($tasks) { } private static function isAlreadyExistsError($errCode, $errMsg) { - if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false) { + if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false || stripos($errMsg, 'existed too recently') !== false) { return true; } if (($errCode === 5 || $errCode === 404) && (stripos($errMsg, 'Requested entity was not found') !== false || stripos($errMsg, 'tombstoned') !== false)) { diff --git a/tests/Api/TaskQueue/PushQueueTest.php b/tests/Api/TaskQueue/PushQueueTest.php index 2901d6ab..b011b47b 100644 --- a/tests/Api/TaskQueue/PushQueueTest.php +++ b/tests/Api/TaskQueue/PushQueueTest.php @@ -278,4 +278,90 @@ public function testPushQueueTooManyTasksError() { $this->apiProxyMock->verify(); } + public function testCloudTasksV2BatchCreateTasksViaGetResult() { + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function batchCreateTasks($req) { + return new class { + public function isDone() { return true; } + public function pollUntilComplete() { return true; } + public function operationFailed() { return false; } + public function getMetadata() { + return new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata(); + } + public function getResult() { + $t1 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/task-1'); + $t2 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/task-2'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse()) + ->setTasks([$t1, $t2]); + } + }; + } + public function close() {} + }; + }); + + $task1 = new PushTask('/url1', ['k' => 'v1'], ['name' => 'task-1']); + $task2 = new PushTask('/url2', ['k' => 'v2'], ['name' => 'task-2']); + $queue = new PushQueue('default'); + $names = $queue->addTasks([$task1, $task2]); + $this->assertEquals(['task-1', 'task-2'], $names); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2BatchCreateTasksFailedRequestsAlreadyExists() { + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function batchCreateTasks($req) { + return new class { + public function isDone() { return true; } + public function pollUntilComplete() { return true; } + public function operationFailed() { return false; } + public function getMetadata() { + $st = (new \Google\Rpc\Status()) + ->setCode(6) + ->setMessage('The task cannot be created because a task with this name existed too recently'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata()) + ->setFailedRequests([1 => $st]); + } + public function getResult() { + $t1 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/task-1'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse()) + ->setTasks([$t1]); + } + }; + } + public function close() {} + }; + }); + + $this->expectException('\Google\AppEngine\Api\TaskQueue\TaskAlreadyExistsException'); + $task1 = new PushTask('/url1'); + $task2 = new PushTask('/url2', [], ['name' => 'dup-task']); + $queue = new PushQueue('default'); + $queue->addTasks([$task1, $task2]); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + } From 24c8f447ee838db62912cddfbb45613c660f555d Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Tue, 6 Oct 2026 10:06:30 +0000 Subject: [PATCH 36/37] test(taskqueue): add Cloud Tasks V2 stubs and unit tests for CI --- tests/Api/TaskQueue/PushQueueTest.php | 223 ++++++++++++++++++++++++++ 1 file changed, 223 insertions(+) diff --git a/tests/Api/TaskQueue/PushQueueTest.php b/tests/Api/TaskQueue/PushQueueTest.php index b011b47b..5990a63a 100644 --- a/tests/Api/TaskQueue/PushQueueTest.php +++ b/tests/Api/TaskQueue/PushQueueTest.php @@ -278,7 +278,186 @@ public function testPushQueueTooManyTasksError() { $this->apiProxyMock->verify(); } + private static function ensureCloudTasksV2StubsLoaded() { + if (!class_exists('\Google\Protobuf\Timestamp')) { + eval(' + namespace Google\Protobuf; + class Timestamp { + private $seconds = 0; + public function setSeconds($s) { $this->seconds = $s; return $this; } + public function getSeconds() { return $this->seconds; } + } + '); + } + if (!class_exists('\Google\Rpc\Status')) { + eval(' + namespace Google\Rpc; + class Status { + private $code = 0; + private $message = ""; + public function setCode($c) { $this->code = $c; return $this; } + public function getCode() { return $this->code; } + public function setMessage($m) { $this->message = $m; return $this; } + public function getMessage() { return $this->message; } + } + '); + } + if (!class_exists('\Google\ApiCore\ApiException')) { + eval(' + namespace Google\ApiCore; + class ApiException extends \Exception { + private $status; + public function __construct($message, $code, $status = "") { + parent::__construct($message, $code); + $this->status = $status; + } + public function getStatus() { return $this->status; } + } + '); + } + if (!class_exists('\Google\Cloud\Tasks\V2\Task')) { + eval(' + namespace Google\Cloud\Tasks\V2; + class AppEngineHttpRequest { + private $relativeUri = "/"; + private $httpMethod = 1; + private $headers; + private $body = ""; + public function __construct() { $this->headers = new \ArrayObject(); } + public function setRelativeUri($u) { $this->relativeUri = $u; return $this; } + public function getRelativeUri() { return $this->relativeUri; } + public function setHttpMethod($m) { $this->httpMethod = $m; return $this; } + public function getHttpMethod() { return $this->httpMethod; } + public function getHeaders() { return $this->headers; } + public function setBody($b) { $this->body = $b; return $this; } + public function getBody() { return $this->body; } + } + class Task { + private $name = ""; + private $appEngineHttpRequest = null; + private $scheduleTime = null; + public function setName($n) { $this->name = $n; return $this; } + public function getName() { return $this->name; } + public function setAppEngineHttpRequest($r) { $this->appEngineHttpRequest = $r; return $this; } + public function getAppEngineHttpRequest() { return $this->appEngineHttpRequest; } + public function setScheduleTime($t) { $this->scheduleTime = $t; return $this; } + public function getScheduleTime() { return $this->scheduleTime; } + } + class CreateTaskRequest { + private $parent = ""; + private $task = null; + public function setParent($p) { $this->parent = $p; return $this; } + public function getParent() { return $this->parent; } + public function setTask($t) { $this->task = $t; return $this; } + public function getTask() { return $this->task; } + } + class BatchCreateTasksRequest { + private $parent = ""; + private $requests = []; + public function setParent($p) { $this->parent = $p; return $this; } + public function getParent() { return $this->parent; } + public function setRequests($r) { $this->requests = $r; return $this; } + public function getRequests() { return $this->requests; } + } + class BatchCreateTasksResponse { + private $tasks = []; + public function setTasks($t) { $this->tasks = $t; return $this; } + public function getTasks() { return $this->tasks; } + } + class BatchCreateTasksMetadata { + private $failedRequests = []; + public function setFailedRequests($f) { $this->failedRequests = $f; return $this; } + public function getFailedRequests() { return $this->failedRequests; } + } + '); + } + if (!class_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient')) { + eval(' + namespace Google\Cloud\Tasks\V2\Client; + class CloudTasksClient { + public function createTask($req) {} + public function batchCreateTasks($req) {} + public function close() {} + } + '); + } + } + + public function testCloudTasksV2SingleCreateTask() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + $capturedReq = null; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$capturedReq) { + return new class($capturedReq) { + private $capturedReqRef; + public function __construct(&$ref) { $this->capturedReqRef = &$ref; } + public function createTask($req) { + $this->capturedReqRef = $req; + return (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/single-1'); + } + public function close() {} + }; + }); + + $task = new PushTask('/worker/push', ['foo' => 'bar'], [ + 'name' => 'single-1', + 'delay_seconds' => 30, + ]); + $queue = new PushQueue('default'); + $names = $queue->addTasks([$task]); + $this->assertEquals(['single-1'], $names); + $this->assertNotNull($capturedReq); + $this->assertEquals( + 'projects/test-proj/locations/us-central1/queues/default', + $capturedReq->getParent()); + $ctTask = $capturedReq->getTask(); + $this->assertEquals( + 'projects/test-proj/locations/us-central1/queues/default/tasks/single-1', + $ctTask->getName()); + $this->assertEquals('/worker/push', $ctTask->getAppEngineHttpRequest()->getRelativeUri()); + $this->assertEquals('foo=bar', $ctTask->getAppEngineHttpRequest()->getBody()); + $this->assertNotNull($ctTask->getScheduleTime()); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2SingleCreateTaskAlreadyExists() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function createTask($req) { + throw new \Google\ApiCore\ApiException('Task already exists', 409, 'ALREADY_EXISTS'); + } + public function close() {} + }; + }); + + $this->expectException('\Google\AppEngine\Api\TaskQueue\TaskAlreadyExistsException'); + $task = new PushTask('/worker/push', [], ['name' => 'dup-single']); + $queue = new PushQueue('default'); + $queue->addTasks([$task]); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + public function testCloudTasksV2BatchCreateTasksViaGetResult() { + self::ensureCloudTasksV2StubsLoaded(); putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); putenv('GOOGLE_CLOUD_PROJECT=test-proj'); putenv('LOCATION_ID=us-central1'); @@ -321,6 +500,7 @@ public function close() {} } public function testCloudTasksV2BatchCreateTasksFailedRequestsAlreadyExists() { + self::ensureCloudTasksV2StubsLoaded(); putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); putenv('GOOGLE_CLOUD_PROJECT=test-proj'); putenv('LOCATION_ID=us-central1'); @@ -364,4 +544,47 @@ public function close() {} } } + public function testCloudTasksV2BatchCreateTasksFailedRequestsUnknownQueue() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function batchCreateTasks($req) { + return new class { + public function isDone() { return true; } + public function pollUntilComplete() { return true; } + public function operationFailed() { return false; } + public function getMetadata() { + $st = (new \Google\Rpc\Status()) + ->setCode(5) + ->setMessage('Queue does not exist'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata()) + ->setFailedRequests([0 => $st]); + } + public function getResult() { + return new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse(); + } + }; + } + public function close() {} + }; + }); + + $this->expectException('\Google\AppEngine\Api\TaskQueue\TaskQueueException'); + $this->expectExceptionMessage('Unknown queue'); + $task1 = new PushTask('/url1'); + $task2 = new PushTask('/url2'); + $queue = new PushQueue('missing-queue'); + $queue->addTasks([$task1, $task2]); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + } From 542c755c65362462abff2c3eb48b40075f73b422 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Wed, 7 Oct 2026 12:54:32 +0000 Subject: [PATCH 37/37] fix(taskqueue): address review feedback on Cloud Tasks push queue path - Reuse a single Cloud Tasks client per addTasks() call for batch chunks and for the per-task fallback path instead of creating one per chunk/task. - Preserve fractional delay_seconds by building schedule_time from microtime(true) with seconds + nanos. - Stop mapping NOT_FOUND ("Requested entity was not found") to TaskAlreadyExistsException; Cloud Tasks reports duplicate and tombstoned task names as ALREADY_EXISTS. - Add unit tests for fractional delay, NOT_FOUND mapping and client reuse. --- src/Api/TaskQueue/PushQueue.php | 290 ++++++++++++++------------ tests/Api/TaskQueue/PushQueueTest.php | 124 +++++++++++ 2 files changed, 280 insertions(+), 134 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index 0248fa6e..d7bae85a 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -356,14 +356,29 @@ private function buildCloudTaskObjV2($task, $fullQueueName) { $taskObj->setAppEngineHttpRequest($appEngineReq); if ($task->getDelaySeconds() > 0) { - $ts = new \Google\Protobuf\Timestamp(); - $ts->setSeconds(time() + $task->getDelaySeconds()); - $taskObj->setScheduleTime($ts); + $taskObj->setScheduleTime(self::buildScheduleTime($task->getDelaySeconds())); } return $taskObj; } + /** + * Builds a Timestamp for now + $delaySeconds, preserving fractional seconds. + */ + private static function buildScheduleTime($delaySeconds) { + $scheduleTime = microtime(true) + $delaySeconds; + $seconds = (int) floor($scheduleTime); + $nanos = (int) round(($scheduleTime - $seconds) * 1e9); + if ($nanos >= 1000000000) { + $seconds += 1; + $nanos -= 1000000000; + } + $ts = new \Google\Protobuf\Timestamp(); + $ts->setSeconds($seconds); + $ts->setNanos($nanos); + return $ts; + } + private function buildCloudTaskObjV2beta3($task, $fullQueueName) { $headers = []; $hasContentType = false; @@ -431,83 +446,75 @@ private function buildCloudTaskObjV2beta3($task, $fullQueueName) { $taskObj->setAppEngineHttpRequest($appEngineReq); if ($task->getDelaySeconds() > 0) { - $ts = new \Google\Protobuf\Timestamp(); - $ts->setSeconds(time() + $task->getDelaySeconds()); - $taskObj->setScheduleTime($ts); + $taskObj->setScheduleTime(self::buildScheduleTime($task->getDelaySeconds())); } return $taskObj; } - private function createSingleTaskCloudTasks($task, $fullQueueName) { - if (class_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient')) { - $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); - $client = self::newCloudTasksClient('\Google\Cloud\Tasks\V2\Client\CloudTasksClient'); - $createTaskReq = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) - ->setParent($fullQueueName) - ->setTask($taskObj); - try { - $response = $client->createTask($createTaskReq); - $parts = explode('/', $response->getName()); - return [end($parts)]; - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); - } - throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); - } finally { - $client->close(); + /** + * Returns [clientClass, apiVersion, usesRequestObjects] for the newest Cloud + * Tasks client installed, or null if no Cloud Tasks client is available. + */ + private static function singleTaskClientSpec() { + $candidates = [ + ['\Google\Cloud\Tasks\V2\Client\CloudTasksClient', 'V2', true], + ['\Google\Cloud\Tasks\V2\CloudTasksClient', 'V2', false], + ['\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient', 'V2beta3', true], + ['\Google\Cloud\Tasks\V2beta3\CloudTasksClient', 'V2beta3', false], + ]; + foreach ($candidates as $candidate) { + if (class_exists($candidate[0])) { + return $candidate; } - } elseif (class_exists('\Google\Cloud\Tasks\V2\CloudTasksClient')) { + } + return null; + } + + /** + * Creates a single task. If $client is null, a client is created and closed + * within this call; otherwise the caller owns the client's lifecycle. + */ + private function createSingleTaskCloudTasks($task, $fullQueueName, $client = null) { + $spec = self::singleTaskClientSpec(); + if ($spec === null) { + throw new TaskQueueException('Cloud Tasks Client SDK is not available.'); + } + list($clientClass, $apiVersion, $usesRequestObjects) = $spec; + + if ($apiVersion === 'V2') { $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); - $client = self::newCloudTasksClient('\Google\Cloud\Tasks\V2\CloudTasksClient'); - try { - $response = $client->createTask($fullQueueName, $taskObj); - $parts = explode('/', $response->getName()); - return [end($parts)]; - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); - } - throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); - } finally { - $client->close(); - } - } elseif (class_exists('\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient')) { + $requestClass = '\Google\Cloud\Tasks\V2\CreateTaskRequest'; + } else { $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); - $client = self::newCloudTasksClient('\Google\Cloud\Tasks\V2beta3\Client\CloudTasksClient'); - $createTaskReq = (new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest()) - ->setParent($fullQueueName) - ->setTask($taskObj); - try { + $requestClass = '\Google\Cloud\Tasks\V2beta3\CreateTaskRequest'; + } + + $ownsClient = ($client === null); + if ($ownsClient) { + $client = self::newCloudTasksClient($clientClass); + } + try { + if ($usesRequestObjects) { + $createTaskReq = (new $requestClass()) + ->setParent($fullQueueName) + ->setTask($taskObj); $response = $client->createTask($createTaskReq); - $parts = explode('/', $response->getName()); - return [end($parts)]; - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); - } - throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); - } finally { - $client->close(); - } - } elseif (class_exists('\Google\Cloud\Tasks\V2beta3\CloudTasksClient')) { - $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); - $client = self::newCloudTasksClient('\Google\Cloud\Tasks\V2beta3\CloudTasksClient'); - try { + } else { $response = $client->createTask($fullQueueName, $taskObj); - $parts = explode('/', $response->getName()); - return [end($parts)]; - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); - } - throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); - } finally { + } + $parts = explode('/', $response->getName()); + return [end($parts)]; + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK createTask failed: ' . $e->getMessage()); + } finally { + if ($ownsClient) { $client->close(); } } - throw new TaskQueueException('Cloud Tasks Client SDK is not available.'); } private static function processBatchCreateResponse($response, $chunk, &$names) { @@ -583,40 +590,43 @@ private function addTasksCloudTasks($tasks) { } if ($v2ClientClass !== null) { - foreach ($chunks as $chunk) { - $createTaskRequests = []; - foreach ($chunk as $task) { - $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); - $createTaskReq = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) - ->setParent($fullQueueName) - ->setTask($taskObj); - $createTaskRequests[] = $createTaskReq; - } - - $client = self::newCloudTasksClient($v2ClientClass); - try { - if (class_exists('\Google\Cloud\Tasks\V2\BatchCreateTasksRequest')) { - $batchReq = (new \Google\Cloud\Tasks\V2\BatchCreateTasksRequest()) + // Create the client once and reuse it for every chunk. + $client = self::newCloudTasksClient($v2ClientClass); + try { + foreach ($chunks as $chunk) { + $createTaskRequests = []; + foreach ($chunk as $task) { + $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); + $createTaskReq = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) ->setParent($fullQueueName) - ->setRequests($createTaskRequests); - try { - $response = $client->batchCreateTasks($batchReq); - } catch (\TypeError $te) { + ->setTask($taskObj); + $createTaskRequests[] = $createTaskReq; + } + + try { + if (class_exists('\Google\Cloud\Tasks\V2\BatchCreateTasksRequest')) { + $batchReq = (new \Google\Cloud\Tasks\V2\BatchCreateTasksRequest()) + ->setParent($fullQueueName) + ->setRequests($createTaskRequests); + try { + $response = $client->batchCreateTasks($batchReq); + } catch (\TypeError $te) { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + } + } else { $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); } - } else { - $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); - } - self::processBatchCreateResponse($response, $chunk, $names); - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + self::processBatchCreateResponse($response, $chunk, $names); + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK batchCreate failed: ' . $e->getMessage()); } - throw new TaskQueueException('Cloud Tasks Client SDK batchCreate failed: ' . $e->getMessage()); - } finally { - $client->close(); } + } finally { + $client->close(); } return $names; } @@ -630,60 +640,72 @@ private function addTasksCloudTasks($tasks) { } if ($betaClientClass !== null) { - foreach ($chunks as $chunk) { - $createTaskRequests = []; - foreach ($chunk as $task) { - $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); - $createTaskReq = new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest(); - $createTaskReq->setParent($fullQueueName); - $createTaskReq->setTask($taskObj); - $createTaskRequests[] = $createTaskReq; - } + // Create the client once and reuse it for every chunk. + $client = self::newCloudTasksClient($betaClientClass); + try { + foreach ($chunks as $chunk) { + $createTaskRequests = []; + foreach ($chunk as $task) { + $taskObj = $this->buildCloudTaskObjV2beta3($task, $fullQueueName); + $createTaskReq = new \Google\Cloud\Tasks\V2beta3\CreateTaskRequest(); + $createTaskReq->setParent($fullQueueName); + $createTaskReq->setTask($taskObj); + $createTaskRequests[] = $createTaskReq; + } - $client = self::newCloudTasksClient($betaClientClass); - try { - if (class_exists('\Google\Cloud\Tasks\V2beta3\BatchCreateTasksRequest')) { - $batchReq = (new \Google\Cloud\Tasks\V2beta3\BatchCreateTasksRequest()) - ->setParent($fullQueueName) - ->setRequests($createTaskRequests); - try { - $response = $client->batchCreateTasks($batchReq); - } catch (\TypeError $te) { + try { + if (class_exists('\Google\Cloud\Tasks\V2beta3\BatchCreateTasksRequest')) { + $batchReq = (new \Google\Cloud\Tasks\V2beta3\BatchCreateTasksRequest()) + ->setParent($fullQueueName) + ->setRequests($createTaskRequests); + try { + $response = $client->batchCreateTasks($batchReq); + } catch (\TypeError $te) { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + } + } else { $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); } - } else { - $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); - } - self::processBatchCreateResponse($response, $chunk, $names); - } catch (\Google\ApiCore\ApiException $e) { - if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { - throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + self::processBatchCreateResponse($response, $chunk, $names); + } catch (\Google\ApiCore\ApiException $e) { + if ($e->getStatus() === 'ALREADY_EXISTS' || $e->getCode() === 409 || self::isAlreadyExistsError($e->getCode(), $e->getMessage())) { + throw new TaskAlreadyExistsException('Task exists already: ' . $e->getMessage()); + } + throw new TaskQueueException('Cloud Tasks Client SDK batchCreate failed: ' . $e->getMessage()); } - throw new TaskQueueException('Cloud Tasks Client SDK batchCreate failed: ' . $e->getMessage()); - } finally { - $client->close(); } + } finally { + $client->close(); } return $names; } // Fallback: If native batchCreateTasks is not available in the installed Cloud Tasks SDK, - // enqueue tasks individually using createTask. - foreach ($tasks as $task) { - $res = $this->createSingleTaskCloudTasks($task, $fullQueueName); - $names[] = $res[0]; + // enqueue tasks individually using createTask over a single shared client. + $spec = self::singleTaskClientSpec(); + if ($spec === null) { + throw new TaskQueueException('Cloud Tasks Client SDK is not available.'); + } + $client = self::newCloudTasksClient($spec[0]); + try { + foreach ($tasks as $task) { + $res = $this->createSingleTaskCloudTasks($task, $fullQueueName, $client); + $names[] = $res[0]; + } + } finally { + $client->close(); } return $names; } private static function isAlreadyExistsError($errCode, $errMsg) { - if ($errCode === 6 || $errCode === 409 || stripos($errMsg, 'already exists') !== false || stripos($errMsg, 'existed too recently') !== false) { - return true; - } - if (($errCode === 5 || $errCode === 404) && (stripos($errMsg, 'Requested entity was not found') !== false || stripos($errMsg, 'tombstoned') !== false)) { - return true; - } - return false; + // Cloud Tasks reports duplicate and tombstoned task names as ALREADY_EXISTS. + // NOT_FOUND is intentionally not mapped here so that missing queues, + // projects or locations surface as TaskQueueException. + return $errCode === 6 || $errCode === 409 || + stripos($errMsg, 'already exists') !== false || + stripos($errMsg, 'existed too recently') !== false || + stripos($errMsg, 'tombstoned') !== false; } } diff --git a/tests/Api/TaskQueue/PushQueueTest.php b/tests/Api/TaskQueue/PushQueueTest.php index 5990a63a..2ade7d0d 100644 --- a/tests/Api/TaskQueue/PushQueueTest.php +++ b/tests/Api/TaskQueue/PushQueueTest.php @@ -284,8 +284,11 @@ private static function ensureCloudTasksV2StubsLoaded() { namespace Google\Protobuf; class Timestamp { private $seconds = 0; + private $nanos = 0; public function setSeconds($s) { $this->seconds = $s; return $this; } public function getSeconds() { return $this->seconds; } + public function setNanos($n) { $this->nanos = $n; return $this; } + public function getNanos() { return $this->nanos; } } '); } @@ -587,4 +590,125 @@ public function close() {} } } + public function testCloudTasksV2SingleCreateTaskFractionalDelay() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + $capturedReq = null; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$capturedReq) { + return new class($capturedReq) { + private $capturedReqRef; + public function __construct(&$ref) { $this->capturedReqRef = &$ref; } + public function createTask($req) { + $this->capturedReqRef = $req; + return (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/frac-1'); + } + public function close() {} + }; + }); + + MockMicrotime::expect(12345.25); + $task = new PushTask('/worker/push', [], [ + 'name' => 'frac-1', + 'delay_seconds' => 0.5, + ]); + $queue = new PushQueue('default'); + $queue->addTasks([$task]); + + $scheduleTime = $capturedReq->getTask()->getScheduleTime(); + $this->assertEquals(12345, $scheduleTime->getSeconds()); + $this->assertEquals(750000000, $scheduleTime->getNanos()); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2SingleCreateTaskNotFoundIsNotAlreadyExists() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function createTask($req) { + throw new \Google\ApiCore\ApiException('Requested entity was not found.', 5, 'NOT_FOUND'); + } + public function close() {} + }; + }); + + $task = new PushTask('/worker/push', [], ['name' => 'nf-1']); + $queue = new PushQueue('default'); + try { + $queue->addTasks([$task]); + $this->fail('Expected TaskQueueException'); + } catch (TaskQueueException $e) { + $this->assertNotInstanceOf(TaskAlreadyExistsException::class, $e); + $this->assertStringContainsString('Requested entity was not found', $e->getMessage()); + } + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2BatchCreateTasksUsesSingleClient() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + $counts = ['created' => 0, 'closed' => 0]; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$counts) { + $counts['created']++; + return new class($counts) { + private $countsRef; + public function __construct(&$ref) { $this->countsRef = &$ref; } + public function batchCreateTasks($req) { + return new class { + public function isDone() { return true; } + public function pollUntilComplete() { return true; } + public function operationFailed() { return false; } + public function getMetadata() { + return new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata(); + } + public function getResult() { + $t1 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/a'); + $t2 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/b'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse()) + ->setTasks([$t1, $t2]); + } + }; + } + public function close() { $this->countsRef['closed']++; } + }; + }); + + $queue = new PushQueue('default'); + $names = $queue->addTasks([ + new PushTask('/a', [], ['name' => 'a']), + new PushTask('/b', [], ['name' => 'b']), + ]); + $this->assertEquals(['a', 'b'], $names); + $this->assertEquals(1, $counts['created']); + $this->assertEquals(1, $counts['closed']); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + }