ClaimedTaskRunner.php
1 month ago
Cli.php
1 month ago
CronCommand.php
1 month ago
DaemonRunner.php
1 month ago
ExecutionLimitOverride.php
1 month ago
ScheduledTaskResolver.php
1 month ago
ScheduledTasksLister.php
1 month ago
TaskAdder.php
1 month ago
TaskCanceller.php
1 month ago
TaskRunner.php
1 month ago
TaskTrigger.php
1 month ago
WorkerTypesCatalog.php
1 month ago
index.php
1 month ago
ClaimedTaskRunner.php
174 lines
| 1 | <?php declare(strict_types = 1); |
| 2 | |
| 3 | namespace MailPoet\Cron\CliCommands; |
| 4 | |
| 5 | if (!defined('ABSPATH')) exit; |
| 6 | |
| 7 | |
| 8 | use MailPoet\Cron\CronHelper; |
| 9 | use MailPoet\Cron\CronWorkerInterface; |
| 10 | use MailPoet\Entities\ScheduledTaskEntity; |
| 11 | use MailPoet\Newsletter\Sending\ScheduledTasksRepository; |
| 12 | use MailPoetVendor\Carbon\Carbon; |
| 13 | use RuntimeException; |
| 14 | use Throwable; |
| 15 | |
| 16 | /** |
| 17 | * Claims a single scheduled-task row for a WP-CLI process and drives it through one worker run. |
| 18 | * |
| 19 | * A claimed row carries status = STATUS_CLI, which is invisible to every daemon-side query, so the web |
| 20 | * daemon can neither pick it up nor reschedule it while the CLI run owns it. On success the row is |
| 21 | * completed (with a meta.cli breadcrumb merged after the worker's last write); on a partial run or |
| 22 | * failure it is handed back to the site cron (scheduled, due now); if requirements are not met it is |
| 23 | * removed. |
| 24 | * |
| 25 | * See doc/wp-cli-cron-commands.md ("CLI execution: the cli status") for the full rationale — why a |
| 26 | * dedicated status, the meta/inProgress placement, and how hard-killed (zombie) claims are handled. |
| 27 | */ |
| 28 | class ClaimedTaskRunner { |
| 29 | private ScheduledTasksRepository $scheduledTasksRepository; |
| 30 | |
| 31 | private ExecutionLimitOverride $executionLimitOverride; |
| 32 | |
| 33 | public function __construct( |
| 34 | ScheduledTasksRepository $scheduledTasksRepository, |
| 35 | ExecutionLimitOverride $executionLimitOverride |
| 36 | ) { |
| 37 | $this->scheduledTasksRepository = $scheduledTasksRepository; |
| 38 | $this->executionLimitOverride = $executionLimitOverride; |
| 39 | } |
| 40 | |
| 41 | /** |
| 42 | * Creates a fresh row already claimed (status STATUS_CLI), due now, and flushes it. No meta and no |
| 43 | * inProgress: the status alone hides the row, and workers may overwrite meta mid-run (see the doc). |
| 44 | * Public so a test can inspect the DB state after the claim. |
| 45 | */ |
| 46 | public function claimNew(string $type, int $priority = ScheduledTaskEntity::PRIORITY_LOW): ScheduledTaskEntity { |
| 47 | $task = new ScheduledTaskEntity(); |
| 48 | $task->setType($type); |
| 49 | $task->setStatus(ScheduledTaskEntity::STATUS_CLI); |
| 50 | $task->setPriority($priority); |
| 51 | $task->setScheduledAt(Carbon::now()->millisecond(0)); |
| 52 | $this->scheduledTasksRepository->persist($task); |
| 53 | $this->scheduledTasksRepository->flush(); |
| 54 | return $task; |
| 55 | } |
| 56 | |
| 57 | /** |
| 58 | * Atomically claims an EXISTING row (scheduled/paused, already resolved by the caller) by transitioning |
| 59 | * it to STATUS_CLI in a single guarded UPDATE. The row's meta is preserved; only the status changes. |
| 60 | * Returns false when the row was no longer claimable (a concurrent CLI run or the site daemon got it |
| 61 | * first), so the caller can skip it instead of double-processing. |
| 62 | */ |
| 63 | public function claimExisting(ScheduledTaskEntity $task): bool { |
| 64 | return $this->scheduledTasksRepository->claimAsCli($task); |
| 65 | } |
| 66 | |
| 67 | /** |
| 68 | * Drives one already-claimed row through a single worker run. The claim is always resolved on a |
| 69 | * non-completing path so the row never stays in 'cli' unless the process dies: on success it is |
| 70 | * completed; on partial work, a failed prepare, a hit execution limit, or an error it is handed back to |
| 71 | * the site cron; when requirements are not met it is removed. Public so a test can drive a fake worker |
| 72 | * through each path directly. |
| 73 | * |
| 74 | * $timeout caps the worker's own execution-limit checks (in seconds); null lifts the cap so the worker |
| 75 | * runs to completion. A worker that hits the cap is handed back gracefully and reported via |
| 76 | * limit_reached rather than as a failure. |
| 77 | * |
| 78 | * @return array{completed: bool, limit_reached: bool, message: string} |
| 79 | */ |
| 80 | public function run(CronWorkerInterface $worker, ScheduledTaskEntity $task, ?int $timeout = null): array { |
| 81 | // Capture the claim marker at the start of the run. Claim and run always happen back-to-back on |
| 82 | // the same instance, so this is the pid/started_at of the run that owns the row. |
| 83 | $marker = [ |
| 84 | 'pid' => getmypid(), |
| 85 | 'started_at' => Carbon::now()->toIso8601String(), |
| 86 | ]; |
| 87 | |
| 88 | $requirementsMet = false; |
| 89 | try { |
| 90 | $requirementsMet = $worker->checkProcessingRequirements(); |
| 91 | } catch (Throwable $e) { |
| 92 | $this->handBack($task); |
| 93 | throw new RuntimeException(sprintf("Task %d (%s) failed while running: %s. It was handed back to the site cron to retry.", $task->getId(), $worker->getTaskType(), $e->getMessage()), 0, $e); |
| 94 | } |
| 95 | |
| 96 | if (!$requirementsMet) { |
| 97 | // Requirements not met: the claimed row would never run, so drop it rather than leave a stuck |
| 98 | // 'cli' task. Mirrors CronWorkerRunner, which removes such tasks. |
| 99 | $this->scheduledTasksRepository->remove($task); |
| 100 | $this->scheduledTasksRepository->flush(); |
| 101 | throw new RuntimeException(sprintf("Requirements for '%s' are not met; the claimed task was removed and nothing ran.", $worker->getTaskType())); |
| 102 | } |
| 103 | |
| 104 | $completed = false; |
| 105 | try { |
| 106 | $worker->init(); |
| 107 | |
| 108 | $this->executionLimitOverride->overrideDuring($timeout, function () use ($worker, $task, &$completed): void { |
| 109 | // Mirror CronWorkerRunner: a task is only processed after prepare succeeds. Bounce is the one |
| 110 | // standard worker that overrides prepare (it builds subscriber rows and returns false when there |
| 111 | // is nothing to do); when prepare returns false we leave $completed false and hand the row back. |
| 112 | if (!$worker->prepareTaskStrategy($task, microtime(true))) { |
| 113 | return; |
| 114 | } |
| 115 | $completed = (bool)$worker->processTaskStrategy($task, microtime(true)); |
| 116 | }); |
| 117 | } catch (Throwable $e) { |
| 118 | $this->handBack($task); |
| 119 | // A worker that hit the execution limit (when $timeout caps the run) is yielding, not failing: |
| 120 | // hand it back so the site cron continues it, and report it as limit_reached rather than an error. |
| 121 | if ($e->getCode() === CronHelper::DAEMON_EXECUTION_LIMIT_REACHED) { |
| 122 | return [ |
| 123 | 'completed' => false, |
| 124 | 'limit_reached' => true, |
| 125 | 'message' => sprintf('Task %d hit the execution limit; it was handed back to the site cron to continue.', $task->getId()), |
| 126 | ]; |
| 127 | } |
| 128 | throw new RuntimeException(sprintf("Task %d (%s) failed while running: %s. It was handed back to the site cron to retry.", $task->getId(), $worker->getTaskType(), $e->getMessage()), 0, $e); |
| 129 | } |
| 130 | |
| 131 | if ($completed) { |
| 132 | $this->complete($task, $marker); |
| 133 | return [ |
| 134 | 'completed' => true, |
| 135 | 'limit_reached' => false, |
| 136 | 'message' => sprintf('Task %d completed.', $task->getId()), |
| 137 | ]; |
| 138 | } |
| 139 | |
| 140 | $this->handBack($task); |
| 141 | return [ |
| 142 | 'completed' => false, |
| 143 | 'limit_reached' => false, |
| 144 | 'message' => sprintf('Task %d processed partially; it was handed back to the site cron to continue.', $task->getId()), |
| 145 | ]; |
| 146 | } |
| 147 | |
| 148 | /** |
| 149 | * Marks the row completed (STATUS_COMPLETED + processedAt) and stamps meta.cli as the permanent |
| 150 | * "done by CLI" breadcrumb. The merge happens AFTER the worker's last write, so a worker that |
| 151 | * overwrote meta wholesale mid-run cannot clobber the breadcrumb. |
| 152 | * |
| 153 | * @param array{pid: int|false, started_at: string} $marker |
| 154 | */ |
| 155 | private function complete(ScheduledTaskEntity $task, array $marker): void { |
| 156 | $task->setProcessedAt(Carbon::now()->millisecond(0)); |
| 157 | $task->setStatus(ScheduledTaskEntity::STATUS_COMPLETED); |
| 158 | $task->setMeta(array_merge($task->getMeta() ?? [], ['cli' => $marker])); |
| 159 | $this->scheduledTasksRepository->persist($task); |
| 160 | $this->scheduledTasksRepository->flush(); |
| 161 | } |
| 162 | |
| 163 | /** |
| 164 | * Hands a not-finished row back to the site cron: STATUS_SCHEDULED, due now. No meta.cli is written |
| 165 | * — the breadcrumb is only for tasks the CLI actually completed. |
| 166 | */ |
| 167 | private function handBack(ScheduledTaskEntity $task): void { |
| 168 | $task->setStatus(ScheduledTaskEntity::STATUS_SCHEDULED); |
| 169 | $task->setScheduledAt(Carbon::now()->millisecond(0)); |
| 170 | $this->scheduledTasksRepository->persist($task); |
| 171 | $this->scheduledTasksRepository->flush(); |
| 172 | } |
| 173 | } |
| 174 |