Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
69 / 69
100.00%
10 / 10
 CRAP
100.00%
1 / 1
Runner
100.00%
69 / 69
100.00%
10 / 10
 27
100.00%
1 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 scoped
100.00%
5 / 5
100.00%
1 / 1
 1
 run
100.00%
11 / 11
100.00%
1 / 1
 6
 next
100.00%
2 / 2
100.00%
1 / 1
 2
 claim
100.00%
4 / 4
100.00%
1 / 1
 2
 process
100.00%
21 / 21
100.00%
1 / 1
 8
 endTransaction
100.00%
4 / 4
100.00%
1 / 1
 3
 retry
100.00%
12 / 12
100.00%
1 / 1
 2
 fail
100.00%
8 / 8
100.00%
1 / 1
 1
 message
100.00%
1 / 1
100.00%
1 / 1
 1
1<?php
2
3declare(strict_types=1);
4
5namespace Cosray\Queue;
6
7use Celema\Container\Container;
8use Celema\Quma\Database;
9use Closure;
10use Cosray\Console\Runtime;
11use Cosray\Contract\Job;
12use Cosray\Exception\JobFailed;
13use Cosray\Exception\RuntimeException;
14use Psr\Log\LoggerInterface as Logger;
15use Throwable;
16
17/**
18 * Runs due jobs one at a time, after a response or from `queue:work`.
19 *
20 * A claimed job holds a lease, so runners in several processes never run
21 * it at once; when the lease ends before the job does, its runner is
22 * presumed dead and the job is due again. A failed attempt is retried
23 * later, with growing delays, until the job's attempts are used up.
24 *
25 * @internal
26 *
27 * @phpstan-type Claimed array{id: int, queue: string, job: string, payload: string, attempts: int, max_attempts: int}
28 */
29final class Runner
30{
31    /** Seconds a claimed job may run before another runner takes it over. */
32    public const int LEASE = 300;
33
34    /** Seconds before the second, third and later attempts. */
35    public const array BACKOFF = [60, 300, 1800, 7200, 43200];
36
37    private const int MAX_ERROR_LENGTH = 4000;
38
39    /**
40     * @param Closure(class-string<Job>): object $resolve builds a job
41     * @param ?Closure(): float $clock seconds, for the time limit
42     */
43    public function __construct(
44        private readonly Database $db,
45        private readonly Logger $logger,
46        private readonly Closure $resolve,
47        private readonly ?Closure $clock = null,
48    ) {}
49
50    /**
51     * A runner that builds jobs in a request-free scope of the container,
52     * made on first use and dropped with the runner.
53     */
54    public static function scoped(Database $db, Logger $logger, Container $container): self
55    {
56        $runtime = null;
57
58        return new self($db, $logger, static function (string $class) use (&$runtime, $container): object {
59            $runtime ??= new Runtime($container);
60
61            return $runtime->get($class);
62        });
63    }
64
65    /**
66     * Runs jobs until none is due or a limit is reached. The time limit is
67     * checked between jobs; a running job is never cut short.
68     *
69     * @param ?string $queue only jobs pushed under this name
70     * @param ?int $jobs at most this many jobs
71     * @param ?float $seconds no new job after this many seconds
72     * @return array{done: int, retried: int, failed: int}
73     */
74    public function run(?string $queue = null, ?int $jobs = null, ?float $seconds = null): array
75    {
76        $clock = $this->clock ?? static fn(): float => hrtime(true) / 1e9;
77        $start = $clock();
78        $counts = ['done' => 0, 'retried' => 0, 'failed' => 0];
79
80        while ($jobs === null || array_sum($counts) < $jobs) {
81            if ($seconds !== null && ($clock() - $start) >= $seconds) {
82                break;
83            }
84
85            $claimed = $this->claim($queue);
86
87            if ($claimed === null) {
88                break;
89            }
90
91            $counts[$this->process($claimed)]++;
92        }
93
94        return $counts;
95    }
96
97    /** Seconds until the next job is due, 0 or less if one is; null without jobs. */
98    public function next(?string $queue = null): ?float
99    {
100        /** @var array{seconds: ?float} $row */
101        $row = $this->db->queue->next($queue === null ? [] : ['queue' => $queue])->one();
102
103        return $row['seconds'];
104    }
105
106    /** @return ?Claimed */
107    private function claim(?string $queue): ?array
108    {
109        $params = ['lease' => self::LEASE];
110
111        if ($queue !== null) {
112            $params['queue'] = $queue;
113        }
114
115        /** @var ?Claimed */
116        return $this->db->queue->claim($params)->first();
117    }
118
119    /**
120     * @param Claimed $claimed
121     * @return 'done'|'retried'|'failed'
122     */
123    private function process(array $claimed): string
124    {
125        if ($claimed['attempts'] > $claimed['max_attempts']) {
126            return $this->fail($claimed, 'Its runner stopped during the last attempt.');
127        }
128
129        $class = $claimed['job'];
130
131        if (!is_a($class, Job::class, true)) {
132            return $this->fail($claimed, "{$class} is not a job class.");
133        }
134
135        $wasInTransaction = $this->db->getConn()->inTransaction();
136
137        try {
138            $job = ($this->resolve)($class);
139
140            if (!$job instanceof Job) {
141                throw new RuntimeException("Resolving {$class} returned " . get_debug_type($job) . '.');
142            }
143
144            $payload = json_decode($claimed['payload'], true, flags: JSON_THROW_ON_ERROR);
145            $job->handle(is_array($payload) ? $payload : []);
146        } catch (JobFailed $e) {
147            $this->endTransaction($wasInTransaction);
148
149            return $this->fail($claimed, self::message($e));
150        } catch (Throwable $e) {
151            $this->endTransaction($wasInTransaction);
152
153            return $this->retry($claimed, self::message($e));
154        }
155
156        if ($this->endTransaction($wasInTransaction)) {
157            return $this->retry($claimed, 'The job left a transaction open.');
158        }
159
160        $this->db->queue->done(['id' => $claimed['id']])->run();
161
162        return 'done';
163    }
164
165    /** Rolls back a transaction the job opened and left open; returns whether there was one. */
166    private function endTransaction(bool $wasInTransaction): bool
167    {
168        if ($wasInTransaction || !$this->db->getConn()->inTransaction()) {
169            return false;
170        }
171
172        $this->db->rollback();
173
174        return true;
175    }
176
177    /**
178     * @param Claimed $claimed
179     * @return 'retried'|'failed'
180     */
181    private function retry(array $claimed, string $error): string
182    {
183        if ($claimed['attempts'] >= $claimed['max_attempts']) {
184            return $this->fail($claimed, $error);
185        }
186
187        $delay = self::BACKOFF[min($claimed['attempts'], count(self::BACKOFF)) - 1];
188        $this->db->queue->retry(['id' => $claimed['id'], 'delay' => $delay, 'error' => $error])->run();
189        $this->logger->warning('Job {id} ({job}) failed attempt {attempt}, next in {delay} seconds: {error}', [
190            'id' => $claimed['id'],
191            'job' => $claimed['job'],
192            'attempt' => $claimed['attempts'],
193            'delay' => $delay,
194            'error' => $error,
195        ]);
196
197        return 'retried';
198    }
199
200    /**
201     * @param Claimed $claimed
202     * @return 'failed'
203     */
204    private function fail(array $claimed, string $error): string
205    {
206        $this->db->queue->fail(['id' => $claimed['id'], 'error' => $error])->run();
207        $this->logger->error('Job {id} ({job}) failed after {attempts} attempts: {error}', [
208            'id' => $claimed['id'],
209            'job' => $claimed['job'],
210            'attempts' => min($claimed['attempts'], $claimed['max_attempts']),
211            'error' => $error,
212        ]);
213
214        return 'failed';
215    }
216
217    private static function message(Throwable $e): string
218    {
219        return mb_substr($e::class . ': ' . $e->getMessage(), 0, self::MAX_ERROR_LENGTH);
220    }
221}