[13.x] Pop managed queue jobs from the cloud-agent instead of SQS (#60659)

kieranbrown · claude · taylorotwell · web-flow · commit 27b256082fb4 · 2026-07-03T15:11:08.000-05:00
* [13.x] Pop managed queue jobs from the cloud-agent instead of SQS Route the cloud Queue decorator's pop() through the in-container cloud-agent's unix-socket runtime API (GET /next) when the underlying connection is an SqsQueue, building an AgentSqsJob instead of receiving from SQS directly. Any non-SQS underlying queue is delegated to as before. The AgentSqsJob still deletes / releases against SQS through the inherited SqsClient — the agent never touches the message — but also reports each terminal outcome back to the agent via POST /result so it can stop heartbeating the message's visibility and accept the next invoke. The agent socket path is configurable via LARAVEL_CLOUD_AGENT_SOCKET, mirroring LARAVEL_CLOUD_LOG_SOCKET. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Report managed queue job outcome to agent instead of mutating SQS AgentSqsJob now calls the grandparent Job::delete()/release() to keep queue:work local state correct without SqsJob's direct SQS calls; the poller owns delete/visibility. reportResultToAgent now sends the release delay. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Always pop from the agent and rename AgentSqsJob to CloudJob The cloud Queue always runs against a managed (SQS) queue, so pop() now polls the agent unconditionally instead of branching on the underlying queue type, inlining the former popFromAgent() helper. The agent socket path is hardcoded to /tmp/cloud-agent.sock, and AgentSqsJob is renamed to CloudJob to match the Cloud\Queue sibling. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Always report processed on delete and strip unused error from result A terminally-failed managed queue job is deleted from SQS just like a successful one (the base fail() routes through delete()), so CloudJob now reports "processed" in both cases rather than a redelivery-triggering "failed". Drops the redundant $failing flag and the fail() override since the base markAsFailed() already runs before delete(). Also strips the unused $error field from the /result report path (the poller never reads it), and switches the agent transport from a raw Guzzle client to the Http facade so the pop loop can be exercised with Http::fake() in tests, matching how FailedJobProvider already talks to the cloud. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Harden CloudJob agent reporting and restore overflow support (#11) Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Harden CloudJob overflow cleanup and agent reporting (#12) Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Make CloudJob agent poller opt-in via config Default the managed-queue connection back to receiving jobs directly from SQS; route pops through the in-container cloud-agent only when agent.enabled is set. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Remove accidentally committed .phpactor.json Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Trust agent attributes and crash the worker on an unreachable agent Drop normalizeAgentAttributes: the dispatcher receives the message from SQS, which always returns ApproximateReceiveCount, so the agent forwards it and we read it straight through like SqsJob::attempts() trusts SQS. Stop rescuing requestNextFromAgent to null. An unreachable (or wedged) agent socket means the agent is not running in the pod, which is unrecoverable, so surface it as an AgentUnreachableException. An AgentAwareLostConnectionDetector (wired for queue:work) treats that as a lost connection, so the worker exits and the pod is restarted instead of idling as if the queue were empty. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Fall back to SQS directly when the cloud-agent is unreachable When reporting an in-flight job's outcome, distinguish an agent that is alive but rejecting the report (HTTP error: retain and let the retry / teardown safety net handle it, since its poller still owns the message) from an agent that has crashed (connection error: its poller can no longer act, so delete()/release() operate on SQS directly via SqsJob's real seams rather than lose the job). Receiving the next job from a crashed agent stays fatal; this only salvages a job already in flight. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Read overflow config directly instead of via getOverflowStorage Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Remove stray whitespace in SqsQueue Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Own CloudJob delete/release instead of no-op overrides Skip SqsJob's SQS calls by flagging the job via the base Job directly, rather than overriding deleteMessageFromSqs/changeMessageVisibilityInSqs/ deleteOverflowPayload to no-ops. The delete/release logic now reads in one place, touching SQS only on the agent-unreachable fallback. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Drop SQS fallback and teardown reporting from CloudJob PHP now only reports a job's outcome to the agent; it never mutates SQS itself. An agent that rejects a report has already finalized the job, so the message redelivers regardless. An unreachable agent propagates as an AgentUnreachableException — the worker treats it as a lost connection and exits (matching the pop side), and SQS redelivers via the visibility timeout once the crashed agent stops heartbeating. Removes CloudJob::fallBackToSqs() and reportToAgent(), and the teardown safety-net call in Queue::finishProcessingJob(). The WorkerStopping listener and shutdown function stay: they flush the terminal queue telemetry event for the last job of every worker run, which is separate from agent reporting. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Make CloudJob report fire-and-trust, dropping report dedup report() no longer tracks reportedStatus or suppresses calls to fake idempotency / a "processed supersedes released" ordering. It reports each outcome as the worker produces it and assumes a returned report was accepted; the agent is the single authority on what ordering of operations is valid. A rejected report (RequestException) now propagates instead of being swallowed, so an outcome the agent cannot apply surfaces rather than being papered over. delete()'s overflow purge is therefore only reached once the agent accepts the "processed" outcome — a thrown report leaves the payload in place for the redelivery. Unreachable still raises AgentUnreachableException; the message redelivers via the visibility timeout regardless. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Inline match into emitted queue event type Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Trim verbose comments to Laravel conventions Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Escalate agent error and malformed responses as unreachable Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Treat any non-200 from GET /next as agent unreachable Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Use Response::ok() helper for the GET /next status check Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Escalate a server error from POST /result as agent unreachable Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Drop the pointless queue parameter from popFromAgent The agent only receives messages, so it has no queue to pop from; each message reports the SQS queue URL it came from. The queue parameter only fed a fallback that never fired, so remove it. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Only pop from the agent for the worker's own queue The cloud-agent sidecar long-polls a single queue on behalf of one queue:work worker, so it only serves work when that worker is popping the queue it was started for. Gate the agent path on running queue:work and the requested queue matching the worker's --queue; anything else falls back to popping from SQS directly. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Document single-queue agent assumption and cover argv forms The cloud-agent serves a single queue, so a managed worker is expected to run one. Note this on workerQueue() and add tests pinning the space-separated --queue form and the multi-queue fallback to SQS. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Report the receipt handle to the agent with each job outcome SQS reuses a message id across redeliveries, so a result matched on the id alone could let a stale outcome finalize a re-dispatched receive. Echo the receipt handle the agent supplied so it can pin the outcome to the exact receive; it is omitted when absent, preserving the id-only fallback. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * [13.x] Scope the agent result retry to connection failures Retry POST /result only on a ConnectionException, matching the method's docblock and FailedJobProvider::find(): a 4xx/5xx is the agent's authoritative response for the message, not a transient hiccup, so it propagates on the first attempt rather than being re-POSTed three times in the job-completion hot path. Also align workerQueue() with WorkCommand's --queue fallback so an option-less managed worker isn't silently misrouted off the agent, and firm up the test argv save/restore and the processed-event queue attribution coverage. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * formatting --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-authored-by: Taylor Otwell <taylor@laravel.com>
diff --git a/src/Illuminate/Foundation/Cloud/AgentAwareLostConnectionDetector.php b/src/Illuminate/Foundation/Cloud/AgentAwareLostConnectionDetector.php
@@ -0,0 +1,32 @@
+<?php
+
+namespace Illuminate\Foundation\Cloud;
+
+use Illuminate\Contracts\Database\LostConnectionDetector;
+use Throwable;
+
+/**
+ * Treats an unreachable cloud-agent runtime socket as a lost connection so the
+ * worker exits and the pod restarts — the only way to recover an agent that
+ * has died in-pod. Every other exception is delegated to parent instance.
+ */
+class AgentAwareLostConnectionDetector implements LostConnectionDetector
+{
+    /**
+     * Create a new detector instance.
+     */
+    public function __construct(
+        protected LostConnectionDetector $detector,
+    ) {
+        //
+    }
+
+    /**
+     * Determine if the given exception was caused by a lost connection.
+     */
+    public function causedByLostConnection(Throwable $e): bool
+    {
+        return $e instanceof AgentUnreachableException
+            || $this->detector->causedByLostConnection($e);
+    }
+}
diff --git a/src/Illuminate/Foundation/Cloud/AgentUnreachableException.php b/src/Illuminate/Foundation/Cloud/AgentUnreachableException.php
@@ -0,0 +1,10 @@
+<?php
+
+namespace Illuminate\Foundation\Cloud;
+
+use RuntimeException;
+
+class AgentUnreachableException extends RuntimeException
+{
+    //
+}
diff --git a/src/Illuminate/Foundation/Cloud/CloudJob.php b/src/Illuminate/Foundation/Cloud/CloudJob.php
@@ -0,0 +1,70 @@
+<?php
+
+namespace Illuminate\Foundation\Cloud;
+
+use Aws\Sqs\SqsClient;
+use Illuminate\Container\Container;
+use Illuminate\Queue\Jobs\Job;
+use Illuminate\Queue\Jobs\SqsJob;
+
+class CloudJob extends SqsJob
+{
+    /**
+     * Create a new job instance.
+     *
+     * @param  array  $job
+     * @param  string  $connectionName
+     * @param  string  $queue
+     * @param  callable(string, int|null): void  $reporter
+     * @param  array  $overflowStorage
+     */
+    public function __construct(
+        Container $container,
+        SqsClient $sqs,
+        array $job,
+        $connectionName,
+        $queue,
+        protected $reporter,
+        array $overflowStorage = [],
+    ) {
+        parent::__construct($container, $sqs, $job, $connectionName, $queue, $overflowStorage);
+    }
+
+    /**
+     * Delete the job from the queue.
+     *
+     * @return void
+     */
+    public function delete()
+    {
+        // Skip SQS deletion so SQS DeleteMessage is left to the poller...
+        Job::delete();
+
+        $this->report('processed');
+
+        // Only reached once the agent has accepted the outcome (report() throws otherwise)...
+        $this->deleteOverflowPayload();
+    }
+
+    /**
+     * Release the job back into the queue after (n) seconds.
+     *
+     * @param  int  $delay
+     * @return void
+     */
+    public function release($delay = 0)
+    {
+        // Skip SQS deletion so SQS release is left to the poller...
+        Job::release($delay);
+
+        $this->report('released', delay: $delay);
+    }
+
+    /**
+     * Report the job's outcome to the agent, which owns the SQS operation.
+     */
+    protected function report(string $status, ?int $delay = null): void
+    {
+        ($this->reporter)($status, $delay);
+    }
+}
diff --git a/src/Illuminate/Foundation/Cloud/Queue.php b/src/Illuminate/Foundation/Cloud/Queue.php
@@ -5,8 +5,13 @@
 use Carbon\CarbonImmutable;
 use Illuminate\Contracts\Queue\ClearableQueue;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
+use Illuminate\Foundation\Application;
+use Illuminate\Http\Client\ConnectionException;
+use Illuminate\Http\Client\RequestException;
+use Illuminate\Support\Facades\Http;
 use Illuminate\Support\Str;
 use Illuminate\Support\Traits\ForwardsCalls;
+use Symfony\Component\Console\Input\ArgvInput;
 
 class Queue implements QueueContract, ClearableQueue
 {
@@ -37,6 +42,7 @@ class Queue implements QueueContract, ClearableQueue
      * Create a new Queue instance.
      */
     public function __construct(
+        protected Application $app,
         protected QueueContract $queue,
         protected Events $events,
         protected array $config,
@@ -185,20 +191,144 @@ public function bulk($jobs, $data = '', $queue = null)
     /**
      * Pop the next job off of the queue.
      *
+     * Jobs come straight from SQS unless the cloud-agent is enabled.
+     *
+     * When cloud-agent is enabled, we long-poll the agent's runtime socket instead.
+     *
      * @param  string|null  $queue
      * @return \Illuminate\Contracts\Queue\Job|null
      */
     public function pop($queue = null)
     {
         $this->finishProcessingJob();
 
-        $job = $this->queue->pop(...func_get_args());
+        $job = $this->usesAgent($queue)
+            ? $this->popFromAgent()
+            : $this->queue->pop(...func_get_args());
 
         $this->startProcessingJob($queue, $job);
 
         return $job;
     }
 
+    /**
+     * Long-poll the cloud-agent's runtime socket and wrap the next message in a CloudJob.
+     *
+     * @return \Illuminate\Foundation\Cloud\CloudJob|null
+     */
+    protected function popFromAgent()
+    {
+        $data = $this->requestNextJobFromAgent();
+
+        if (! (is_array($data) && is_string($messageId = $data['messageId'] ?? null) && $messageId !== '')) {
+            return null;
+        }
+
+        // Coerce a non-string handle to null so a malformed response degrades gracefully...
+        $receiptHandle = is_string($handle = $data['receiptHandle'] ?? null) ? $handle : null;
+
+        return new CloudJob(
+            $this->queue->getContainer(),
+            $this->queue->getSqs(),
+            [
+                'MessageId' => $messageId,
+                'ReceiptHandle' => $receiptHandle,
+                'Body' => is_string($body = $data['body'] ?? null) ? $body : '',
+                'Attributes' => $data['attributes'] ?? [],
+            ],
+            $this->queue->getConnectionName(),
+            $data['queueUrl'] ?? null,
+            fn (string $status, ?int $delay) => $this->reportJobStatusToAgent(
+                $messageId, $receiptHandle, $status, $delay
+            ),
+            $this->config['connection']['overflow'] ?? [],
+        );
+    }
+
+    /**
+     * Long-poll the agent's runtime socket (GET /next) for the next job.
+     *
+     * @throws \Illuminate\Foundation\Cloud\AgentUnreachableException
+     */
+    protected function requestNextJobFromAgent(): ?array
+    {
+        try {
+            $response = $this->agentRequest()
+                ->timeout(65)
+                ->get('/next');
+        } catch (ConnectionException $e) {
+            throw new AgentUnreachableException(
+                'The Laravel Cloud agent runtime socket is unreachable.', previous: $e
+            );
+        }
+
+        if ($response->status() === 204) {
+            return null;
+        }
+
+        if (! $response->ok()) {
+            throw new AgentUnreachableException(
+                "The Laravel Cloud agent returned HTTP {$response->status()} from GET /next."
+            );
+        }
+
+        if (! is_array($data = $response->json())) {
+            throw new AgentUnreachableException(
+                'The Laravel Cloud agent returned a non-array body from GET /next.'
+            );
+        }
+
+        return $data;
+    }
+
+    /**
+     * Report a job's terminal outcome back to the agent (POST /result). so it
+     *
+     * @throws \Illuminate\Http\Client\RequestException
+     * @throws \Illuminate\Foundation\Cloud\AgentUnreachableException
+     */
+    protected function reportJobStatusToAgent(string $messageId, ?string $receiptHandle, string $status, ?int $delay = null): void
+    {
+        try {
+            $this->agentRequest()
+                ->timeout(10)
+                ->throw()
+                ->retry(3, 100, fn ($exception) => $exception instanceof ConnectionException)
+                ->post('/result', array_filter([
+                    'messageId' => $messageId,
+                    'receiptHandle' => $receiptHandle,
+                    'status' => $status,
+                    'delay' => $delay,
+                ], fn ($value) => $value !== null));
+        } catch (ConnectionException $e) {
+            throw new AgentUnreachableException(
+                'The Laravel Cloud agent runtime socket is unreachable.', previous: $e
+            );
+        } catch (RequestException $e) {
+            if ($e->response->serverError()) {
+                throw new AgentUnreachableException(
+                    "The Laravel Cloud agent returned HTTP {$e->response->status()} from POST /result.", previous: $e
+                );
+            }
+
+            throw $e;
+        }
+    }
+
+    /**
+     * Get a pending HTTP request bound to the agent's Unix runtime socket.
+     *
+     * @return \Illuminate\Http\Client\PendingRequest
+     */
+    protected function agentRequest()
+    {
+        return Http::baseUrl('http://localhost')->withOptions([
+            'curl' => [
+                CURLOPT_UNIX_SOCKET_PATH => $this->config['agent']['socket'] ?? '/tmp/cloud-agent.sock',
+            ],
+        ]);
+    }
+
     /**
      * Delete all of the jobs from the queue.
      *
@@ -210,6 +340,29 @@ public function clear($queue)
         return $this->queue->clear(...func_get_args());
     }
 
+    /**
+     * Determine whether the next job should be received from the in-container cloud-agent.
+     *
+     * @param  string|null  $queue
+     */
+    protected function usesAgent($queue = null): bool
+    {
+        return ($this->config['agent']['enabled'] ?? false)
+            && $this->app->runningConsoleCommand('queue:work')
+            && $this->queue->getQueue($queue) === $this->queue->getQueue($this->workerQueue());
+    }
+
+    /**
+     * Get the queue the running queue:work worker is processing.
+     *
+     * @return string
+     */
+    protected function workerQueue()
+    {
+        return (new ArgvInput)->getParameterOption('--queue')
+            ?: ($this->config['queue'] ?? 'default');
+    }
+
     /**
      * Get the connection name for the queue.
      *
diff --git a/src/Illuminate/Foundation/Cloud/QueueConnector.php b/src/Illuminate/Foundation/Cloud/QueueConnector.php
@@ -5,6 +5,7 @@
 use Aws\CommandInterface;
 use Aws\Exception\AwsException;
 use Aws\Sqs\SqsClient;
+use Illuminate\Contracts\Database\LostConnectionDetector;
 use Illuminate\Foundation\Application;
 use Illuminate\Queue\Connectors\ConnectorInterface;
 use Illuminate\Queue\Events\JobQueued;
@@ -39,6 +40,7 @@ public function connect(array $config): Queue
         $underlying = $this->connector->connect($config['connection']);
 
         $queue = new Queue(
+            $this->app,
             $underlying,
             $this->app[Events::class],
             $config,
@@ -101,6 +103,12 @@ protected function configureWorker(Queue $queue): void
         Worker::$restartable = false;
         Worker::$pausable = false;
 
+        // Exit the worker (and restart the pod) when the agent socket is unreachable...
+        $this->app->extend(
+            LostConnectionDetector::class,
+            fn ($detector) => new AgentAwareLostConnectionDetector($detector),
+        );
+
         $this->app['events']->listen(fn (WorkerStopping $event) => match ($event->reason) {
             WorkerStopReason::TimedOut => $queue->finishProcessingJob(default: 'released'),
             default => $queue->finishProcessingJob(),
diff --git a/src/Illuminate/Queue/Jobs/SqsJob.php b/src/Illuminate/Queue/Jobs/SqsJob.php
@@ -67,6 +67,17 @@ public function release($delay = 0)
     {
         parent::release($delay);
 
+        $this->changeMessageVisibilityInSqs($delay);
+    }
+
+    /**
+     * Reset the SQS message's visibility timeout so it becomes available again.
+     *
+     * @param  int  $delay
+     * @return void
+     */
+    protected function changeMessageVisibilityInSqs($delay)
+    {
         $this->sqs->changeMessageVisibility([
             'QueueUrl' => $this->queue,
             'ReceiptHandle' => $this->job['ReceiptHandle'],
@@ -83,10 +94,30 @@ public function delete()
     {
         parent::delete();
 
+        $this->deleteMessageFromSqs();
+
+        $this->deleteOverflowPayload();
+    }
+
+    /**
+     * Delete the message from the SQS queue.
+     *
+     * @return void
+     */
+    protected function deleteMessageFromSqs()
+    {
         $this->sqs->deleteMessage([
             'QueueUrl' => $this->queue, 'ReceiptHandle' => $this->job['ReceiptHandle'],
         ]);
+    }
 
+    /**
+     * Delete the offloaded overflow payload from the cache, if applicable.
+     *
+     * @return void
+     */
+    protected function deleteOverflowPayload()
+    {
         if (Arr::get($this->overflowStorage, 'delete_after_processing') &&
             $pointer = $this->overflowPointer()) {
             $this->overflowStore()->forget($pointer);
diff --git a/tests/Foundation/Cloud/QueueTest.php b/tests/Foundation/Cloud/QueueTest.php
