diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d5bc3b..72c980f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## 1.0.0 under development +- Bug #19: Fix a claimed run being handed out a second time when it waits for a free process slot under `--max-processes`, or is handed out behind a task slower than the reclaim window; the window now starts when the run starts rather than when it was claimed (@KalimeroMK) +- New #19: Record what became of a run: it is claimed as it is handed out and recorded as done, failed, skipped or cancelled once its outcome is known, so a run nobody started is given back to the schedule instead of passing for one that happened, and a claim left by a process that is gone is handed out again. Added `TaskStatus` and `Schedule::reclaimAfter()` (@KalimeroMK) - Enh #17: Keep a checkpoint record per task instead of one record holding every run emitted at a due time, so that a stored value stays the same size however many tasks share that time and a run costs a single write rather than a write of everything emitted before it (@KalimeroMK) - Enh #16: Generate a task identifier with a 64-bit hash instead of a 32-bit one, so that a task added to a schedule cannot be taken for one the checkpoint holds under the same identifier (@KalimeroMK) - Bug #16: Fix a run being skipped, or one already made being repeated, when a task is added to or removed from the schedule while the checkpoint holds the runs of a shared due time; the checkpoint now records which tasks already ran by their identifier instead of by their place in the schedule (@KalimeroMK) diff --git a/README.md b/README.md index 6693f17..fca811f 100644 --- a/README.md +++ b/README.md @@ -208,6 +208,31 @@ $schedule = (new Schedule()) ->task(...); ``` +## What became of a run + +A run is claimed in the checkpoint as it is handed out and recorded as `done`, `failed`, +`skipped` or `cancelled` — the cases of `TaskStatus` — once its outcome is known. A run the +scheduler never gets to, because it is shutting down, is given back to the schedule and stays +due rather than passing for a run that happened. + +A claim is left behind when the process holding it dies. Every claim is taken back after a +window, set with `reclaimAfter()` and a day by default: + +```php +$schedule->reclaimAfter(900); // or a DateInterval +``` + +The window has to outlast a run, counted from the moment it starts. Under `--max-processes` a +due run waits for a free slot first, and that wait is not charged against the window. Taking a +run back while its process still means to make it runs the task a second time. + +A claim made by a process of this host is taken back sooner than that, as soon as the process +is gone, which needs `ext-posix`. Asking about the process only brings the window forward and +never puts it off, so a claim whose process is still there is taken back once the window runs +out as well. A process being alive is no promise that the run is: a pid is given to something +else once the process it belonged to is gone, and a claim waiting on one of those would stand +forever. + ## Task identifiers Every task gets an identifier generated from the cron expression or the interval of its diff --git a/src/Generator/Checkpoint.php b/src/Generator/Checkpoint.php index 7e98061..c20e030 100644 --- a/src/Generator/Checkpoint.php +++ b/src/Generator/Checkpoint.php @@ -4,69 +4,95 @@ namespace Yiisoft\Scheduler\Generator; +use DateInterval; use DateTimeImmutable; use Psr\SimpleCache\CacheInterface; use Throwable; use Yiisoft\Mutex\MutexInterface; use Yiisoft\Scheduler\Exception\RuntimeException; +use Yiisoft\Scheduler\TaskStatus; use function array_keys; +use function array_pad; +use function explode; +use function function_exists; +use function gethostname; +use function getmypid; use function hash; use function is_array; use function is_string; +use function posix_get_last_error; +use function posix_kill; use function sprintf; /** * The persisted-or-not position of a schedule: which runs were already emitted. * - * Each task keeps the time of its own last emitted run, and the schedule keeps the time it was - * last evaluated to the end. A task resumes after its own record; one that has none — newly - * added, or added while the schedule was down — resumes from the position of the schedule. + * Each task keeps the time of its own last emitted run and what became of it; the schedule keeps + * the time it was last evaluated to the end, which is where a task with no record starts. One + * record per task keeps a stored value the same size however many tasks share a due time. * - * A record per task keeps a stored value the same size however many tasks share a due time, and - * a run one write instead of a write of everything emitted so far. + * A run is claimed as it is handed out and resolved once its outcome is known, so another process + * reads a run in flight as taken and one that was never started can be given back. + * + * A claim carries the process that made it, and the run is handed out again once that process + * is gone or the reclaim window runs out, whichever comes first. A process of another host, or + * of a host without ext-posix, cannot be asked about, and a live one is no promise that the run + * is still going either, since its pid may since have been given to something else. The window + * is what bounds both, and asking about the process only brings it forward. * * @internal + * + * @psalm-type Record = array{DateTimeImmutable, TaskStatus, string|null, DateTimeImmutable|null} */ final class Checkpoint { private const CACHE_KEY_PREFIX = 'yii_schedule_checkpoint_'; + /** The errno of kill(2) for a process that exists but is not ours to signal. */ + private const EPERM = 1; + + /** + * How long to wait for the schedule lock before giving up on recording what became of a run. + * Long enough to outlast another process's read and write, short enough that a scheduler is + * not held forever by a lock whose holder is gone. + */ + private const WRITE_LOCK_TIMEOUT_SECONDS = 10; + private ?DateTimeImmutable $time = null; - /** @var array Task id to the time of its last emitted run. */ - private array $lastRuns = []; + /** @var array Run, outcome, and the owner and moment of a claim. */ + private array $records = []; private bool $locked = false; private int $revision = 0; - /** - * @param list $taskIds Ids of the tasks of the schedule, whose records are read back. - */ + /** @var list Ids of the tasks whose records are read back. */ + private array $taskIds = []; + public function __construct( private readonly string $scheduleName, - private readonly array $taskIds = [], private readonly ?CacheInterface $cache = null, private readonly ?MutexInterface $mutex = null, - ) { - $this->load(); - } + ) {} /** - * Acquires the schedule mutex, if any. Returns false when another process holds it. + * Acquires the schedule mutex, if any. Returns false when another process holds it. The state + * is re-read each time: another process may have advanced it while this one stood by. * - * The persisted position is re-read on every acquisition: another process may have advanced - * it while this one was standing by, and replaying its runs would run them twice. + * @param list $taskIds Ids of the tasks of the schedule as it stands now, so a task + * added after the scheduler was built has its record read back. */ - public function acquire(): bool + public function acquire(array $taskIds): bool { if (null !== $this->mutex && !$this->mutex->acquire(0)) { return false; } + $this->taskIds = $taskIds; + $this->locked = null !== $this->mutex; - // The caller only releases what it was told it acquired, so a position that cannot be - // read must not leave the mutex behind: nobody else could run the schedule until this - // process is gone. + // The caller only releases what it was told it acquired, so state that cannot be read + // must not leave the mutex behind. try { $this->load(); } catch (Throwable $error) { @@ -78,10 +104,7 @@ public function acquire(): bool return true; } - /** - * Changes whenever the position is replaced by one read from the shared state, so a caller - * caching work derived from the position knows to recompute it. - */ + /** Changes when the state is replaced by one read back, so cached work can be recomputed. */ public function revision(): int { return $this->revision; @@ -96,36 +119,51 @@ public function release(): void } } - /** The time the schedule was last evaluated to the end: the seed of a task with no record. */ + /** The time the schedule was last evaluated to the end: where a task with no record starts. */ public function time(): ?DateTimeImmutable { return $this->time; } - public function lastRun(string $taskId): ?DateTimeImmutable + /** + * @psalm-return Record|null + * + * @return array{DateTimeImmutable, TaskStatus, string|null, DateTimeImmutable|null}|null + */ + public function record(string $taskId): ?array { - return $this->lastRuns[$taskId] ?? null; + return $this->records[$taskId] ?? null; } /** - * Records the run as emitted. Never moves backwards: a catch-up reports the runs it skips - * here too, and an earlier time would make the ones after them look unemitted. + * Writes the record of a task. Claiming, resolving and giving back all come through here, and + * each is the one writer of that record, so none has to guard against the others. */ - public function save(string $taskId, DateTimeImmutable $time): void - { - $recorded = $this->lastRuns[$taskId] ?? null; - - if (null !== $recorded && $time <= $recorded) { - return; - } - - $this->lastRuns[$taskId] = $time; + public function write( + string $taskId, + DateTimeImmutable $time, + TaskStatus $status, + ?string $owner = null, + ?DateTimeImmutable $at = null, + ): void { + $claimed = TaskStatus::Claimed === $status; + $owner = $claimed ? $owner ?? self::owner() : null; + $at = $claimed ? $at ?? $time : null; + + $this->records[$taskId] = [$time, $status, $owner, $at]; if (null === $this->cache) { return; } - if (!$this->cache->set($this->taskKey($taskId), $time->format('U.u'))) { + $state = [ + 'time' => $time->format('U.u'), + 'status' => $status->value, + 'owner' => $owner, + 'at' => $at?->format('U.u'), + ]; + + if (!$this->cache->set($this->taskKey($taskId), $state)) { throw new RuntimeException(sprintf( 'Unable to persist the checkpoint of task "%s" of schedule "%s".', $taskId, @@ -135,14 +173,31 @@ public function save(string $taskId, DateTimeImmutable $time): void } /** - * Records that the schedule was evaluated to the end: every run due at or before this time is - * emitted, so a task with no record of its own resumes from here. Only an uninterrupted tick - * reaches it, which is what leaves the runs of a process that died partway still due. + * Restarts the reclaim window of a claimed run at the moment the run actually begins. + * + * A run claimed as it is handed out may then wait for a free process slot. Charging that wait + * against the window would hand the run out a second time while the process holding it is + * still about to start it, so the window counts from here instead of from the claim. + * + * Anything but this process's own claim of this very run is left alone: the record has moved + * on, and whoever moved it is the one that owns it now. + * + * @return bool Whether the run is still this process's to start. False means the wait for a + * free slot outlasted the claim and another process has taken the run over, so starting it + * here would run it twice. + */ + public function markStarted(string $taskId, DateTimeImmutable $time, DateTimeImmutable $at): bool + { + return $this->exclusively(fn(): bool => $this->startUnderLock($taskId, $time, $at)); + } + + /** + * Records that the schedule was evaluated to the end, which is what lets a cron setup see the + * window (lastTick, now] as still to cover. Only an uninterrupted tick reaches it. */ public function markTick(DateTimeImmutable $now): void { - // A process re-reading an older state must not drag the position back over runs emitted - // in between. + // A process re-reading an older state must not drag the position back. if (null !== $this->time && $now <= $this->time) { return; } @@ -168,6 +223,213 @@ public function markInitialPosition(DateTimeImmutable $now): void $this->writePosition($now); } + /** + * Where the task resumes: after its recorded run, or just before it when that run is to be + * handed out again: given back unstarted, or claimed by a process that is gone or has been + * holding it longer than the schedule waits. + */ + public function resumePoint(string $taskId, DateTimeImmutable $now, DateInterval $reclaimAfter): ?DateTimeImmutable + { + $record = $this->records[$taskId] ?? null; + + if (null === $record) { + return null; + } + + [$time, $status] = $record; + + if (TaskStatus::Released === $status || self::isReclaimable($record, $now, $reclaimAfter)) { + return $time->modify('-1 microsecond'); + } + + return $time; + } + + /** + * Whether any record is a claim there is no longer any point in waiting on. A heap built while + * a claim held has to be rebuilt once that stops being true, or the run it stands for is never + * handed out by a scheduler that keeps running. + */ + public function hasReclaimableClaim(DateTimeImmutable $now, DateInterval $reclaimAfter): bool + { + foreach ($this->records as $record) { + if (self::isReclaimable($record, $now, $reclaimAfter)) { + return true; + } + } + + return false; + } + + /** + * Records what became of a claimed run, if the state still holds this process's claim of it. + * + * An outcome arrives long after the claim was made, and by then the task may have been taken + * over: the claim ran out of window, or the run after it was already handed out and finished + * elsewhere. Writing the outcome anyway would move the record behind a run that is over, and + * hand that run out again. + */ + public function resolve(string $taskId, DateTimeImmutable $time, TaskStatus $status): void + { + $this->exclusively(function () use ($taskId, $time, $status): void { + $this->resolveUnderLock($taskId, $time, $status); + }); + } + + /** + * Hands a claimed run back, leaving it due. A later run than the one on record is dropped the + * same way, so giving back several leaves the task resuming before the earliest of them. + */ + public function giveBack(string $taskId, DateTimeImmutable $time): void + { + $this->exclusively(function () use ($taskId, $time): void { + $this->giveBackUnderLock($taskId, $time); + }); + } + + /** + * Carries out a read, a decision made on what it says, and the write that follows from it, + * with nobody else writing the state in between. + * + * Reading the record back is not enough on its own: between that read and the write, another + * scheduler can finish a later run of the same task, and the write then puts the record back + * where that run started. The claim is made under the schedule lock, and so is what is + * recorded about it afterwards. + * + * A tick holds the lock from the moment it starts generating, and a sequential run is + * resolved inside that, so the lock is taken here only when this process is not already + * holding it: a mutex is not reentrant and taking it twice would wait on this process. + * + * @template T + * + * @param callable(): T $operation + * + * @return T + */ + private function exclusively(callable $operation): mixed + { + if ($this->locked || null === $this->mutex) { + return $operation(); + } + + // Waiting, unlike the tick, which moves on to let another process run the schedule: what + // became of a run that already happened has to be recorded, and whoever holds the lock + // holds it for a read and a write. + if (!$this->mutex->acquire(self::WRITE_LOCK_TIMEOUT_SECONDS)) { + throw new RuntimeException(sprintf( + 'Unable to record the outcome of a run of schedule "%s": the schedule lock was held' + . ' for longer than %d seconds.', + $this->scheduleName, + self::WRITE_LOCK_TIMEOUT_SECONDS, + )); + } + + $this->locked = true; + + try { + return $operation(); + } finally { + $this->locked = false; + $this->mutex->release(); + } + } + + private function startUnderLock(string $taskId, DateTimeImmutable $time, DateTimeImmutable $at): bool + { + $claim = $this->ourClaim($taskId); + + // At or before the record, not exactly on it: one tick of catch-up claims every missed + // run of a task before it starts any of them, and they share the one record, which + // stands for the newest. A run at or before it is one this process claimed itself; a + // later one is a run it never claimed. + if (null === $claim || $time > $claim[0]) { + return false; + } + + // Only the newest claim carries the window. An earlier run of the same catch-up starts + // under the record as it is rather than moving it back to its own time, which is what + // would hand the runs between the two out again. + if ($time == $claim[0]) { + $this->write($taskId, $time, TaskStatus::Claimed, self::owner(), $at); + } + + return true; + } + + private function resolveUnderLock(string $taskId, DateTimeImmutable $time, TaskStatus $status): void + { + $claim = $this->ourClaim($taskId); + + // Only the newest run claimed is the one the record stands for; the outcome of an older + // one of the same catch-up says nothing about where the task has got to. + if (null === $claim || $time != $claim[0]) { + return; + } + + $this->write($taskId, $time, $status); + } + + private function giveBackUnderLock(string $taskId, DateTimeImmutable $time): void + { + $claim = $this->ourClaim($taskId); + + if (null === $claim || $time > $claim[0]) { + return; + } + + $this->write($taskId, $time, TaskStatus::Released); + } + + /** + * The record of a task if the shared state still holds it as a claim of this process. + * + * It is read back first: the one in hand was read when the run was claimed, and what has + * happened to it since is the whole question. Another process may have taken the task over, + * because the claim ran out of window or because the run after it was handed out and + * finished. Nothing may then be recorded about it from here. + * + * The time on the record is the newest run claimed, which is not necessarily the one being + * reported on: a catch-up claims every missed run of a task before any of them is started, + * and they all share the one record. + * + * @psalm-return Record|null + * + * @return array{DateTimeImmutable, TaskStatus, string|null, DateTimeImmutable|null}|null + */ + private function ourClaim(string $taskId): ?array + { + $this->reread($taskId); + + $record = $this->records[$taskId] ?? null; + + if (null === $record) { + return null; + } + + [, $status, $owner] = $record; + + return TaskStatus::Claimed === $status && self::owner() === $owner ? $record : null; + } + + /** Replaces the record of one task with what the shared state holds for it now. */ + private function reread(string $taskId): void + { + if (null === $this->cache) { + return; + } + + /** @var mixed $stored */ + $stored = $this->cache->get($this->taskKey($taskId)); + + if (null === $stored) { + unset($this->records[$taskId]); + + return; + } + + $this->records[$taskId] = $this->readRecord($stored, $taskId); + } + private function writePosition(DateTimeImmutable $now): void { $this->time = $now; @@ -183,6 +445,58 @@ private function writePosition(DateTimeImmutable $now): void } } + /** + * @psalm-param Record $record + */ + private static function isReclaimable(array $record, DateTimeImmutable $now, DateInterval $reclaimAfter): bool + { + [, $status, $owner, $at] = $record; + + if (TaskStatus::Claimed !== $status) { + return false; + } + + return (null !== $owner && self::isGone($owner)) + || (null !== $at && $at->add($reclaimAfter) <= $now); + } + + /** The process a claim is made by, as "host/pid". */ + private static function owner(): string + { + return self::host() . '/' . (getmypid() ?: 0); + } + + /** + * Only a process of this host can be asked about, and only with ext-posix. Anything else + * reads as still there, and is left to the reclaim window: a run handed out twice before + * the window is out is worse than one picked up late. + */ + private static function host(): string + { + return gethostname() ?: 'unknown'; + } + + private static function isGone(string $owner): bool + { + [$host, $pid] = array_pad(explode('/', $owner, 2), 2, ''); + + // A payload holding anything but an exact pid reads as a process still there. + if ($host !== self::host() || !function_exists('posix_kill') || (string) (int) $pid !== $pid) { + return false; + } + + if ((int) $pid <= 0) { + return false; + } + + if (posix_kill((int) $pid, 0)) { + return false; + } + + // The signal is refused for a process of another user, which is still a process. + return self::EPERM !== posix_get_last_error(); + } + private function load(): void { if (null === $this->cache) { @@ -203,23 +517,23 @@ private function load(): void } $time = $this->readPosition($values[$this->cacheKey()] ?? null); - $lastRuns = []; + $records = []; foreach ($taskKeys as $key => $taskId) { /** @var mixed $stored */ $stored = $values[$key] ?? null; if (null !== $stored) { - $lastRuns[$taskId] = $this->readRunTime($stored, $taskId); + $records[$taskId] = $this->readRecord($stored, $taskId); } } - if ($time == $this->time && $lastRuns == $this->lastRuns) { + if ($time == $this->time && $records == $this->records) { return; } $this->time = $time; - $this->lastRuns = $lastRuns; + $this->records = $records; ++$this->revision; } @@ -230,8 +544,9 @@ private function readPosition(mixed $state): ?DateTimeImmutable return null; } - // Either earlier format carries what was emitted at its time alongside it; only the time - // is read, so an upgrade resumes after those runs without replaying them. + // A position written before the runs were recorded per task carries the place, or the + // ids, of the runs emitted at its time alongside it. Only the time is read, so those + // runs are left as emitted and an upgrade resumes after them without replaying any. if (!is_array($state) || !isset($state['time']) || !is_string($state['time'])) { throw new RuntimeException( sprintf('The checkpoint of schedule "%s" holds an unexpected payload.', $this->scheduleName), @@ -241,19 +556,69 @@ private function readPosition(mixed $state): ?DateTimeImmutable return DateTimeImmutable::createFromFormat('U.u', $state['time']) ?: null; } - private function readRunTime(mixed $stored, string $taskId): DateTimeImmutable + /** + * @psalm-return Record + * + * @return array{DateTimeImmutable, TaskStatus, string|null, DateTimeImmutable|null} + */ + private function readRecord(mixed $stored, string $taskId): array { + // A record written before a run carried what became of it holds the time on its own. + // Such a run is over either way, so it reads as done and the task resumes after it. + $status = TaskStatus::Done; + $owner = null; + $at = null; + + if (is_array($stored)) { + if (!isset($stored['time'], $stored['status']) || !is_string($stored['status'])) { + throw new RuntimeException($this->unexpectedTaskPayload($taskId)); + } + + $status = TaskStatus::tryFrom($stored['status']); + + if (null === $status) { + throw new RuntimeException($this->unexpectedTaskPayload($taskId)); + } + + /** @var mixed $holder */ + $holder = $stored['owner'] ?? null; + + if (null !== $holder && !is_string($holder)) { + throw new RuntimeException($this->unexpectedTaskPayload($taskId)); + } + + /** @var mixed $claimedAt */ + $claimedAt = $stored['at'] ?? null; + + if (null !== $claimedAt) { + $at = is_string($claimedAt) ? DateTimeImmutable::createFromFormat('U.u', $claimedAt) : false; + + if (!$at instanceof DateTimeImmutable) { + throw new RuntimeException($this->unexpectedTaskPayload($taskId)); + } + } + + $owner = $holder; + /** @var mixed $stored */ + $stored = $stored['time']; + } + $time = is_string($stored) ? DateTimeImmutable::createFromFormat('U.u', $stored) : false; if (!$time instanceof DateTimeImmutable) { - throw new RuntimeException(sprintf( - 'The checkpoint of task "%s" of schedule "%s" holds an unexpected payload.', - $taskId, - $this->scheduleName, - )); + throw new RuntimeException($this->unexpectedTaskPayload($taskId)); } - return $time; + return [$time, $status, $owner, $at]; + } + + private function unexpectedTaskPayload(string $taskId): string + { + return sprintf( + 'The checkpoint of task "%s" of schedule "%s" holds an unexpected payload.', + $taskId, + $this->scheduleName, + ); } private function cacheKey(): string @@ -262,10 +627,13 @@ private function cacheKey(): string } /** - * Both halves of the identity are hashed rather than spelled out: withId() takes any string - * and a schedule name is as free, while PSR-16 reserves `{}()/\@:` and guarantees keys only - * up to 64 characters. Two fixed-width hashes hold a task key at 57 of them however long the - * schedule and the task are named, and still leave the tasks of one schedule a shared prefix. + * The key a task keeps its record under. + * + * Both halves of the identity are hashed rather than spelled out: {@see + * \Yiisoft\Scheduler\RecurringTask::withId()} takes any string and a schedule name is as + * free, while PSR-16 reserves `{}()/\@:` in a key and guarantees keys only up to 64 + * characters. Two fixed-width hashes hold a task key at 57 of them however long the schedule + * and the task are named, and still leave the tasks of one schedule a shared prefix. */ private function taskKey(string $taskId): string { diff --git a/src/Generator/TaskGenerator.php b/src/Generator/TaskGenerator.php index 566d1eb..1a6141f 100644 --- a/src/Generator/TaskGenerator.php +++ b/src/Generator/TaskGenerator.php @@ -10,6 +10,7 @@ use Yiisoft\Scheduler\RecurringTask; use Yiisoft\Scheduler\Schedule; use Yiisoft\Scheduler\TaskContext; +use Yiisoft\Scheduler\TaskStatus; use function array_map; use function count; @@ -34,12 +35,7 @@ public function __construct( private readonly Schedule $schedule, private readonly ClockInterface $clock, ) { - $this->checkpoint = new Checkpoint( - $schedule->getName(), - array_map(static fn(RecurringTask $task): string => $task->getId(), $schedule->tasks()), - $schedule->getState(), - $schedule->getMutex(), - ); + $this->checkpoint = new Checkpoint($schedule->getName(), $schedule->getState(), $schedule->getMutex()); } public function getSchedule(): Schedule @@ -56,7 +52,7 @@ public function getTasks(): iterable { $now = $this->clock->now(); - if (!$this->checkpoint->acquire()) { + if (!$this->checkpoint->acquire($this->taskIds())) { return; } @@ -85,18 +81,21 @@ public function getTasks(): iterable } if (!$yield) { - $this->checkpoint->save($task->getId(), $time); + $this->checkpoint->write($task->getId(), $time, TaskStatus::Skipped); continue; } $context = new TaskContext($this->schedule->getName(), $task->getId(), $task->getTrigger(), $time, $nextTime); - yield $context => $task; + // Taken rather than due for anyone reading the schedule while this run is in + // flight. resolve() says what became of it, release() gives it back. + // + // Stamped as the claim is made, not as the tick began: the runs of one tick are + // handed out one after another, and the ones behind a slow task would otherwise + // start out with their reclaim window already part spent. + $this->checkpoint->write($task->getId(), $time, TaskStatus::Claimed, at: $this->clock->now()); - // Reached whether the task succeeded or failed: the consumer keeps a failure to - // itself until the schedule is drained, so a failed run is recorded like any - // other instead of being emitted again by the next process. - $this->checkpoint->save($task->getId(), $time); + yield $context => $task; } // Reached only by a tick that drained, so a process that died partway leaves the @@ -116,7 +115,7 @@ public function getTasks(): iterable */ public function getNextRunDate(): ?DateTimeImmutable { - if (!$this->checkpoint->acquire()) { + if (!$this->checkpoint->acquire($this->taskIds())) { return null; } @@ -129,11 +128,58 @@ public function getNextRunDate(): ?DateTimeImmutable } } + /** Records what became of a claimed run, once the consumer knows. */ + public function resolve(TaskContext $context, TaskStatus $status): void + { + $this->checkpoint->resolve($context->taskId, $context->triggeredAt, $status); + } + + /** + * Restarts the reclaim window of a claimed run at the moment the run actually begins, for a + * consumer that starts a run later than it took it. + * + * @return bool Whether the run is still this process's to start. + */ + public function markStarted(TaskContext $context): bool + { + return $this->checkpoint->markStarted($context->taskId, $context->triggeredAt, $this->clock->now()); + } + + /** + * Gives a claimed run back, leaving it due. For a run known not to have been started at all, + * as opposed to one whose outcome is merely unknown. + */ + public function release(TaskContext $context): void + { + $this->checkpoint->giveBack($context->taskId, $context->triggeredAt); + + // The pending runs held here were worked out when this one had been handed out, and the + // run after it is the only one of the task they still carry. Dropping them puts the run + // just given back in front of this scheduler too, rather than only in front of the next + // process to read the state. + $this->heap = null; + } + + /** + * @return list + */ + private function taskIds(): array + { + return array_map(static fn(RecurringTask $task): string => $task->getId(), $this->schedule->tasks()); + } + private function heap(): TriggerHeap { - // A position read from the shared state replaces the pending runs: they were computed from - // a position this process believed to be current, and another one has moved on since. - if (null !== $this->heap && $this->heapRevision === $this->checkpoint->revision()) { + $now = $this->clock->now(); + $reclaimAfter = $this->schedule->getReclaimAfter(); + + // State read back replaces the pending runs: another process has moved on since they were + // computed. So does a claim there is no longer any point in waiting on, which changes + // where its task resumes without changing anything that was stored. + if (null !== $this->heap + && $this->heapRevision === $this->checkpoint->revision() + && !$this->checkpoint->hasReclaimableClaim($now, $reclaimAfter) + ) { return $this->heap; } @@ -146,16 +192,16 @@ private function heap(): TriggerHeap // has ever had. Where they are computed from is written down before any of them is // emitted: a tick interrupted before it drains then leaves the runs it never started // behind a position, rather than behind nothing for the next process to seed from. - $position = $this->clock->now(); + $position = $now; $this->checkpoint->markInitialPosition($position); } foreach ($this->schedule->tasks() as $task) { $index = $this->indices[$task->getId()] ??= count($this->indices); - // No record yet — newly added, or left unstarted when a process died — means the + // No record yet — newly added, or added while the schedule was down — means the // position of the schedule. - $seedTime = $this->checkpoint->lastRun($task->getId()) ?? $position; + $seedTime = $this->checkpoint->resumePoint($task->getId(), $now, $reclaimAfter) ?? $position; if (null !== ($nextTime = $task->getTrigger()->getNextRunDate($seedTime))) { $this->heap->insert([$nextTime, $index, $task]); diff --git a/src/Schedule.php b/src/Schedule.php index f0fe468..68bf420 100644 --- a/src/Schedule.php +++ b/src/Schedule.php @@ -5,10 +5,14 @@ namespace Yiisoft\Scheduler; use Psr\SimpleCache\CacheInterface; +use DateInterval; +use DateTimeImmutable; use Yiisoft\Mutex\MutexInterface; use Throwable; use function array_key_exists; +use function is_int; +use function max; use function array_values; use function sprintf; @@ -22,6 +26,7 @@ final class Schedule private ?MutexInterface $mutex = null; private ?CacheInterface $state = null; private bool $onlyLastMissedRun = false; + private ?DateInterval $reclaimAfter = null; /** @var list */ private array $beforeListeners = []; /** @var list */ @@ -89,6 +94,41 @@ public function stateful(CacheInterface $cache): static return $this; } + /** + * Sets how long a claimed run is left to its process before the schedule takes it back. + * Every claim is taken back once the window runs out, including one whose process is still + * there; a gone process of this host is noticed before that with ext-posix. + * + * It has to outlast the work itself. The window starts when the run starts rather than when + * it was claimed, so the wait for a free slot under `--max-processes` is not charged against + * it. Taking a run back while its process still means to make it runs the task twice. + * + * @param DateInterval|int $after The window, as a date interval or a number of seconds. + */ + public function reclaimAfter(DateInterval|int $after): static + { + if (is_int($after)) { + $after = new DateInterval(sprintf('PT%dS', max(0, $after))); + } + + // A window of nothing would make every claim abandoned the moment it is read back, which + // is the double run it is there to prevent. + $epoch = new DateTimeImmutable('@0'); + + if ($epoch->add($after) <= $epoch) { + throw new Exception\LogicException('The reclaim window must be greater than zero.'); + } + + $this->reclaimAfter = $after; + + return $this; + } + + public function getReclaimAfter(): DateInterval + { + return $this->reclaimAfter ??= new DateInterval('P1D'); + } + /** * Whether the schedule checkpoint is persisted. * diff --git a/src/Scheduler.php b/src/Scheduler.php index ab2a734..a4158c2 100644 --- a/src/Scheduler.php +++ b/src/Scheduler.php @@ -20,6 +20,7 @@ use function array_key_exists; use function array_keys; +use function array_slice; use function array_map; use function cli_set_process_title; use function count; @@ -148,7 +149,16 @@ public function tick(bool $concurrent = false, ?int $maxProcesses = null): int ++$count; try { - $this->runTask($task, $context); + $status = $this->runTask($task, $context); + } catch (Throwable $error) { + $status = TaskStatus::Failed; + $failure = $this->firstFailure($failure, $error); + } + + // Recorded after the run, so a checkpoint that cannot be written is + // reported behind the task's own failure rather than instead of it. + try { + $this->resolve($context, $status); } catch (Throwable $error) { $failure = $this->firstFailure($failure, $error); } @@ -182,20 +192,40 @@ public function tick(bool $concurrent = false, ?int $maxProcesses = null): int } } - // Draining the generators above already moved every checkpoint past these runs, so a - // task that fails to start must not abort the ones queued behind it: they would never - // be emitted again. Collect the first failure and report it once the batch is started. + // Draining the generators claimed every one of these runs, so a task that fails to start + // must not abort the ones behind it: their claims would be left with nobody to resolve + // them. A stop is different — there the runs nobody got to are given back above. foreach ($due as $index => [$task, $context]) { try { if (!$this->awaitFreeSlot($maxProcesses)) { + $pending = array_slice($due, $index); + $this->logger->warning( - 'Stopping, {count} due task(s) of this tick are not started.', - ['count' => count($due) - $index], + 'Stopping, {count} due task(s) of this tick are given back to the schedule.', + ['count' => count($pending)], ); + foreach ($pending as [, $unstarted]) { + $this->release($unstarted); + } + break; } + // The claim was made as the generator was drained; the wait above must not + // count against the window that decides when another process may take the run. + // A wait that outlasted the claim is also how the run stops being this + // process's to start, and starting it then would be the second time it ran. + if (!$this->markStarted($context)) { + $this->logger->warning( + 'Task {task} of schedule {schedule} was taken over while it waited for a' + . ' process slot, so it is not started here.', + ['task' => $context->taskId, 'schedule' => $context->scheduleName], + ); + + continue; + } + $this->forkTask($task, $context); } catch (Throwable $error) { $failure = $this->firstFailure($failure, $error); @@ -442,13 +472,32 @@ private function firstFailure(?Throwable $first, Throwable $error): Throwable return $first; } - private function runTask(RecurringTask $task, TaskContext $context): void + /** Runs the task and reports what became of it, leaving the caller to record it. */ + private function runTask(RecurringTask $task, TaskContext $context): TaskStatus { if (!$this->beforeRun($task, $context)) { - return; + return TaskStatus::Cancelled; } - $this->executeTask($task, $context); + return $this->executeTask($task, $context); + } + + /** Records what became of a claimed run with the schedule it belongs to. */ + private function resolve(TaskContext $context, TaskStatus $status): void + { + $this->generators[$context->scheduleName]->resolve($context, $status); + } + + /** Gives a claimed run that was never started back to the schedule, leaving it due. */ + private function release(TaskContext $context): void + { + $this->generators[$context->scheduleName]->release($context); + } + + /** Whether the run is still this process's to start. */ + private function markStarted(TaskContext $context): bool + { + return $this->generators[$context->scheduleName]->markStarted($context); } /** @@ -477,7 +526,7 @@ private function beforeRun(RecurringTask $task, TaskContext $context): bool return true; } - private function executeTask(RecurringTask $task, TaskContext $context): void + private function executeTask(RecurringTask $task, TaskContext $context): TaskStatus { $schedule = $this->generators[$context->scheduleName]->getSchedule(); @@ -499,13 +548,15 @@ private function executeTask(RecurringTask $task, TaskContext $context): void ['id' => $context->taskId, 'schedule' => $context->scheduleName, 'error' => $error->getMessage()], ); - return; + return TaskStatus::Failed; } $this->dispatcher?->dispatch(new PostRunEvent($context, $task, $result)); foreach ($schedule->getAfterListeners() as $listener) { $listener($context, $task, $result); } + + return TaskStatus::Done; } /** @@ -523,10 +574,14 @@ private function forkTask(RecurringTask $task, TaskContext $context): void ['id' => $context->taskId, 'schedule' => $context->scheduleName], ); + $this->resolve($context, TaskStatus::Skipped); + return; } if (!$this->beforeRun($task, $context)) { + $this->resolve($context, TaskStatus::Cancelled); + return; } @@ -545,7 +600,15 @@ private function forkTask(RecurringTask $task, TaskContext $context): void self::unblockChildSignal($blocked); // Forking is refused (e.g. resource limits): run inline rather than drop the run. - $this->executeTask($task, $context); + try { + $status = $this->executeTask($task, $context); + } catch (Throwable $error) { + $this->resolve($context, TaskStatus::Failed); + + throw $error; + } + + $this->resolve($context, $status); return; } @@ -687,6 +750,8 @@ private function handleChildOutcome(RecurringTask $task, TaskContext $context, i $schedule = $this->generators[$context->scheduleName]->getSchedule(); if (pcntl_wifexited($status) && 0 === pcntl_wexitstatus($status)) { + $this->resolve($context, TaskStatus::Done); + // The result cannot cross the process boundary, so listeners see null here. $this->dispatcher?->dispatch(new PostRunEvent($context, $task, null)); foreach ($schedule->getAfterListeners() as $listener) { @@ -696,6 +761,8 @@ private function handleChildOutcome(RecurringTask $task, TaskContext $context, i return; } + $this->resolve($context, TaskStatus::Failed); + $error = new RuntimeException( pcntl_wifexited($status) ? sprintf('The task process exited with code %d.', pcntl_wexitstatus($status)) diff --git a/src/TaskStatus.php b/src/TaskStatus.php new file mode 100644 index 0000000..2e73bf2 --- /dev/null +++ b/src/TaskStatus.php @@ -0,0 +1,33 @@ +assertSame(['added', 'third'], $log->getArrayCopy()); } + #[RequiresPhpExtension('posix')] + #[RequiresPhpExtension('pcntl')] + public function testClaimOfAProcessThatIsGoneIsHandedOutAgain(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $clock = new TestClock(); + $cache = new InMemoryCache(); + + // A pid that has been and is no more, so asking about it is answered rather than guessed. + $pid = pcntl_fork(); + + if (0 === $pid) { + exit(0); + } + + pcntl_waitpid($pid, $status); + + $claimedAt = $clock->now()->modify('+1 minute'); + $this->leaveClaim($cache, $task, $claimedAt, (gethostname() ?: 'unknown') . '/' . $pid); + + $clock->set($claimedAt->modify('+30 seconds')); + + // The window is a year out, so only the process being gone can account for the run + // being handed out again. + $this->tick($cache, $clock, [$task], 31_536_000); + + $this->assertSame(['task'], $log->getArrayCopy()); + } + + public function testClaimOfAnotherHostIsLeftUntilTheReclaimWindowRunsOut(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $clock = new TestClock(); + $cache = new InMemoryCache(); + + $claimedAt = $clock->now()->modify('+1 minute'); + $this->leaveClaim($cache, $task, $claimedAt, 'another-host/1'); + + // A process of another host cannot be asked about, so the run stays taken while the + // window holds. + $clock->set($claimedAt->modify('+30 seconds')); + $this->tick($cache, $clock, [$task], 3600); + + $this->assertSame([], $log->getArrayCopy()); + + // Once the window runs out it is handed out again, whatever host made the claim. + $this->tick($cache, $clock, [$task], 1); + + $this->assertSame(['task'], $log->getArrayCopy()); + } + + public function testResolvingAnEarlierRunDoesNotHandOutTheOnesAfterItAgain(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $claimed = $this->claimThreeMissedRuns($cache, $clock, $task, $generator); + + // The outcomes arrive out of the order the runs were claimed in: the later ones are + // skipped at once because the first is still going, and the first is resolved last, + // when its process is reaped. + $generator->resolve($claimed[2], TaskStatus::Skipped); + $generator->resolve($claimed[1], TaskStatus::Skipped); + $generator->resolve($claimed[0], TaskStatus::Done); + unset($generator); + + $this->tick($cache, $clock, [$task]); + + $this->assertSame([], $log->getArrayCopy()); + } + + public function testGivingBackSeveralRunsLeavesACompletedOneAlone(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $claimed = $this->claimThreeMissedRuns($cache, $clock, $task, $generator); + + // The first run was started and reaped; the stop then gets to the other two. + $generator->resolve($claimed[0], TaskStatus::Done); + $generator->release($claimed[1]); + $generator->release($claimed[2]); + unset($generator); + + $this->tick($cache, $clock, [$task]); + + // The two given back are due again; the one that ran is not. + $this->assertSame(['task', 'task'], $log->getArrayCopy()); + } + + public function testEveryRunOfOneCatchUpMayBeStartedHere(): void + { + // A tick of catch-up claims every missed run of a task before it starts any of them, + // and they share the one record, which stands for the newest. The earlier ones are this + // process's own: refusing them loses the whole backlog but its last run. + $task = static fn() => null; + $claimed = $this->claimThreeMissedRuns($cache, $clock, $task, $generator); + + foreach ($claimed as $context) { + $this->assertTrue( + $generator->markStarted($context), + sprintf('the run of %s is this scheduler\'s own', $context->triggeredAt->format('H:i:s')), + ); + } + } + + public function testARunThisSchedulerNeverClaimedIsNotStartedHere(): void + { + // The other side of the same guard: at or before the record is ours, past it is not. + $task = static fn() => null; + $claimed = $this->claimThreeMissedRuns($cache, $clock, $task, $generator); + $newest = $claimed[2]; + + $later = new TaskContext( + $newest->scheduleName, + $newest->taskId, + $newest->trigger, + $newest->triggeredAt->modify('+1 minute'), + null, + ); + + $this->assertFalse($generator->markStarted($later)); + } + + public function testRunTakenOverWhileItWaitedForASlotIsNotStartedHere(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $clock = new TestClock(); + $cache = new InMemoryCache(); + $generator = new TaskGenerator($this->schedule($cache, [$task], 10), $clock); + $this->assertNotNull($generator->getNextRunDate()); + + $clock->advance('+1 minute'); + $claimed = null; + + foreach ($generator->getTasks() as $context => $ignored) { + $claimed = $context; + } + + $this->assertNotNull($claimed); + + // The run waits for a free process slot for longer than the reclaim window, and another + // scheduler takes it over in the meantime. + $clock->advance('+20 seconds'); + $this->leaveClaim($cache, $task, $clock->now(), 'another-host/1'); + + // Starting it here as well is the second time it runs, so the claim is checked first. + $this->assertFalse($generator->markStarted($claimed)); + + // Nor does the process that lost the run get to say what became of it. + $generator->resolve($claimed, TaskStatus::Done); + $record = $cache->get($this->taskKey($task)); + + $this->assertIsArray($record); + $this->assertSame('claimed', $record['status']); + $this->assertSame('another-host/1', $record['owner']); + } + + public function testLateOutcomeDoesNotMoveTheRecordBehindARunAnotherSchedulerFinished(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $clock = new TestClock(); + $cache = new InMemoryCache(); + + // One scheduler takes the run due at 00:01 and is still running it. + $first = new TaskGenerator($this->schedule($cache, [$task]), $clock); + $this->assertNotNull($first->getNextRunDate()); + + $clock->advance('+1 minute'); + $claimedByFirst = null; + + foreach ($first->getTasks() as $context => $ignored) { + $claimedByFirst = $context; + } + + // Another takes the run due at 00:02 and finishes it while the first is still going. + $clock->advance('+1 minute'); + $second = new TaskGenerator($this->schedule($cache, [$task]), $clock); + $claimedBySecond = null; + + foreach ($second->getTasks() as $context => $ignored) { + $claimedBySecond = $context; + } + + $this->assertNotNull($claimedBySecond); + $second->resolve($claimedBySecond, TaskStatus::Done); + + // Only now does the first scheduler reap its child and report the older outcome. + $this->assertNotNull($claimedByFirst); + $first->resolve($claimedByFirst, TaskStatus::Done); + + // The run of 00:02 is over, so what comes next is 00:03. Reporting an older outcome + // against a record this scheduler last read before the other one moved it would put + // the finished run back in front of everyone. + $third = new TaskGenerator($this->schedule($cache, [$task]), $clock); + + $this->assertEquals($claimedBySecond->triggeredAt->modify('+1 minute'), $third->getNextRunDate()); + } + + public function testRunGivenBackIsStillPendingForTheSchedulerThatGaveItBack(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $clock = new TestClock(); + $cache = new InMemoryCache(); + $generator = new TaskGenerator($this->schedule($cache, [$task]), $clock); + + // The daemon works out the pending runs when it asks what to wait for, and keeps them. + $this->assertNotNull($generator->getNextRunDate()); + + $clock->advance('+1 minute'); + $claimed = null; + + foreach ($generator->getTasks() as $context => $ignored) { + $claimed = $context; + } + + $this->assertNotNull($claimed); + + // A graceful stop hands back the runs it never got to start. + $generator->release($claimed); + + // What was handed back is pending again for this scheduler too, not only for the next + // one to read the state: the pending runs it is holding no longer describe the record. + $this->assertEquals($claimed->triggeredAt, $generator->getNextRunDate()); + } + + public function testClaimExpiringUnderALongLivedSchedulerIsStillHandedOut(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $clock = new TestClock(); + $cache = new InMemoryCache(); + + $claimedAt = $clock->now()->modify('+1 minute'); + $this->leaveClaim($cache, $task, $claimedAt, 'another-host/1'); + + // One scheduler for both ticks, so the pending runs it works out are the cached ones. + $schedule = $this->schedule($cache, [$task], 60); + $scheduler = new Scheduler([$schedule], new CallableTaskHandler(), $clock); + + $clock->set($claimedAt->modify('+30 seconds')); + $scheduler->tick(); + + $this->assertSame([], $log->getArrayCopy()); + + $clock->set($claimedAt->modify('+90 seconds')); + $scheduler->tick(); + + // The claim has run out of window, so its run is handed out alongside the one that came + // due in the meantime, rather than being left behind by the cached heap. + $this->assertSame(['task', 'task'], $log->getArrayCopy()); + } + + public function testClaimGivenBackIsDueAgain(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $cache = new InMemoryCache(); + $clock = new TestClock(); + $schedule = $this->schedule($cache, [$task]); + + (new Scheduler([$schedule], new CallableTaskHandler(), $clock))->tick(); + $clock->advance('+1 minute'); + + $generator = new TaskGenerator($schedule, $clock); + $claimed = null; + + foreach ($generator->getTasks() as $context => $_) { + // The consumer is stopping and will not get to this one. + $claimed = $context; + break; + } + + $this->assertNotNull($claimed); + $generator->release($claimed); + unset($generator); + + // Given back rather than passed off as a run that happened, so it is still due. + $this->tick($cache, $clock, [$task]); + + $this->assertSame(['task'], $log->getArrayCopy()); + } + + public function testClaimOfAProcessThatIsStillThereIsLeftAlone(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $cache = new InMemoryCache(); + $clock = new TestClock(); + $schedule = $this->schedule($cache, [$task]); + + (new Scheduler([$schedule], new CallableTaskHandler(), $clock))->tick(); + $clock->advance('+1 minute'); + + $generator = new TaskGenerator($schedule, $clock); + + foreach ($generator->getTasks() as $context => $_) { + // Handed out and still in flight: the claim is neither resolved nor given back. + break; + } + + // The process that holds it is this one, and the reclaim window has not run out, so the + // run stays taken. Handing it out again here would run the task a second time. + $this->tick($cache, $clock, [$task]); + + $this->assertSame([], $log->getArrayCopy()); + } + + #[RequiresPhpExtension('posix')] + public function testClaimOfAProcessThatIsStillThereIsTakenBackOnceTheWindowRunsOut(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $cache = new InMemoryCache(); + $clock = new TestClock(); + $schedule = $this->schedule($cache, [$task], 10); + + (new Scheduler([$schedule], new CallableTaskHandler(), $clock))->tick(); + $clock->advance('+1 minute'); + + $generator = new TaskGenerator($schedule, $clock); + + foreach ($generator->getTasks() as $context => $_) { + // Handed out and still in flight: the claim is neither resolved nor given back. + break; + } + + // The process that holds the claim is this one, and it is alive. Asking about it only + // ever brings the window forward, so the run is handed out again once the window runs + // out: a pid is given to something else once its process is gone, and a claim left + // waiting on one of those would stand forever. + $clock->advance('+20 seconds'); + $this->tick($cache, $clock, [$task], 10); + + $this->assertSame(['task'], $log->getArrayCopy()); + } + public function testRunsAlreadyRecordedAreNotWrittenToTheCheckpointAgain(): void { $log = new ArrayObject(); @@ -127,10 +510,10 @@ public function testRunsAlreadyRecordedAreNotWrittenToTheCheckpointAgain(): void $this->tick($cache, $clock, [$first, $second, $third]); - // The two tasks that already have a record are picked up after it and write nothing; - // what is written is the record of the run left unstarted and the position of the tick, - // rather than a record per task in the schedule. - $this->assertSame(2, $cache->writes() - $writes); + // The two tasks that already have a record are picked up after it and write nothing. What + // is written is the run taken back: claimed as it is handed out and recorded once it is + // over, plus the position of the tick — not a record per task in the schedule. + $this->assertSame(3, $cache->writes() - $writes); } public function testCheckpointWrittenBeforeRunsWereRecordedByIdResumesWithoutReplaying(): void @@ -195,7 +578,11 @@ public function testEachTaskKeepsARecordOfItsOwn(): void $this->assertSame(['time' => $clock->now()->format('U.u')], $cache->get('yii_schedule_checkpoint_default')); foreach ([$first, $second] as $task) { - $this->assertSame($clock->now()->format('U.u'), $cache->get($this->taskKey($task))); + // A run that is over keeps no owner: nobody is going to report anything about it. + $this->assertSame( + ['time' => $clock->now()->format('U.u'), 'status' => 'done', 'owner' => null, 'at' => null], + $cache->get($this->taskKey($task)), + ); } } @@ -208,17 +595,20 @@ public function testRunLeftUnstartedByAnInterruptedFirstTickIsStillDue(): void $second = static function () use ($log): void { $log[] = 'second'; }; + $third = static function () use ($log): void { + $log[] = 'third'; + }; $cache = new InMemoryCache(); $clock = new TestClock(); // A daemon asks when the schedule is next due before anything has ever run, so this is // the first time the pending runs are computed from a state holding nothing at all. - $generator = new TaskGenerator($this->schedule($cache, [$first, $second]), $clock); + $generator = new TaskGenerator($this->schedule($cache, [$first, $second, $third]), $clock); $this->assertNotNull($generator->getNextRunDate()); - // The tick it wakes up for is killed between the two runs, so it never drains and never - // records that the schedule was evaluated to the end. + // The tick it wakes up for is killed partway, so it never drains and never records that + // the schedule was evaluated to the end. $clock->advance('+1 minute'); $handler = new CallableTaskHandler(); $started = 0; @@ -235,13 +625,14 @@ public function testRunLeftUnstartedByAnInterruptedFirstTickIsStillDue(): void unset($generator); $this->assertSame(['first'], $log->getArrayCopy()); - // The process replacing it has never seen a drained tick, so the run left unstarted is - // due only if the first process wrote down where the schedule began before emitting - // anything. Seeding that task from the now of this process would step over it. + // The last task was never reached, so it is the one with nothing of its own on record: + // the two before it are claims, and a claim is what keeps their runs. It is due only + // because the first process wrote down where the schedule began before emitting + // anything; seeding it from the now of this process would step over its run. $log->exchangeArray([]); - $this->tick($cache, $clock, [$first, $second]); + $this->tick($cache, $clock, [$first, $second, $third]); - $this->assertSame(['second'], $log->getArrayCopy()); + $this->assertSame(['third'], $log->getArrayCopy()); } public function testRecordsFitACacheLimitedToTheKeyLengthPsr16Guarantees(): void @@ -295,9 +686,237 @@ public function testMalformedCheckpointIsRejected(mixed $state): void $this->tick($cache, new TestClock(), [$task]); } + public function testWaitingForAProcessSlotIsNotChargedAgainstTheReclaimWindow(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $cache = new InMemoryCache(); + $clock = new TestClock(); + $schedule = $this->schedule($cache, [$task], 10); + + (new Scheduler([$schedule], new CallableTaskHandler(), $clock))->tick(); + $clock->advance('+1 minute'); + + $generator = new TaskGenerator($schedule, $clock); + $claimed = null; + + foreach ($generator->getTasks() as $context => $_) { + $claimed = $context; + } + + $this->assertNotNull($claimed); + + // Longer than the whole window goes by before a process slot frees up. + $clock->advance('+20 seconds'); + $generator->markStarted($claimed); + + // Five seconds into the run itself, which is well inside the window counted from here. + $clock->advance('+5 seconds'); + $log->exchangeArray([]); + $this->tick($cache, $clock, [$task], 10); + + $this->assertSame([], $log->getArrayCopy()); + } + + public function testARunHandedOutBehindASlowOneKeepsItsWholeReclaimWindow(): void + { + $log = new ArrayObject(); + $first = static function () use ($log): void { + $log[] = 'first'; + }; + $second = static function () use ($log): void { + $log[] = 'second'; + }; + + $cache = new InMemoryCache(); + $clock = new TestClock(); + $schedule = $this->schedule($cache, [$first, $second], 10); + + (new Scheduler([$schedule], new CallableTaskHandler(), $clock))->tick(); + $clock->advance('+1 minute'); + + // Both runs are due at the same minute and are handed out one after the other, with the + // first taking longer than the window to run. + $generator = new TaskGenerator($schedule, $clock); + + $slow = true; + + foreach ($generator->getTasks() as $_) { + if ($slow) { + $clock->advance('+20 seconds'); + $slow = false; + } + } + + unset($generator); + + // Neither run was resolved, so the first one is out of window and taken back. The second + // was claimed twenty seconds later and still has time left on it. + $clock->advance('+5 seconds'); + $log->exchangeArray([]); + $this->tick($cache, $clock, [$first, $second], 10); + + $this->assertSame(['first'], $log->getArrayCopy()); + } + + public function testMarkingAResolvedRunAsStartedLeavesItsRecordAlone(): void + { + $log = new ArrayObject(); + $task = static function () use ($log): void { + $log[] = 'task'; + }; + + $cache = new InMemoryCache(); + $clock = new TestClock(); + $schedule = $this->schedule($cache, [$task], 10); + + (new Scheduler([$schedule], new CallableTaskHandler(), $clock))->tick(); + $clock->advance('+1 minute'); + + $generator = new TaskGenerator($schedule, $clock); + $claimed = null; + + foreach ($generator->getTasks() as $context => $_) { + $claimed = $context; + } + + $this->assertNotNull($claimed); + $generator->resolve($claimed, TaskStatus::Done); + + // A late start of a run that is already over must not put the claim back. + $generator->markStarted($claimed); + + $record = $cache->get($this->taskKey($task)); + + $this->assertIsArray($record); + $this->assertSame('done', $record['status']); + } + + /** + * @param list $tasks + */ + public function testWhatBecameOfARunIsRecordedUnderTheScheduleLock(): void + { + // Rereading the record is not enough on its own: between that read and the write, + // another scheduler can finish a later run of the same task, and the write would then + // put the record back where that run started. + $mutex = new class implements MutexInterface { + public int $acquired = 0; + public int $released = 0; + private bool $held = false; + + public function acquire(int $timeout = 0): bool + { + if ($this->held) { + return false; + } + + ++$this->acquired; + + return $this->held = true; + } + + public function release(): void + { + ++$this->released; + $this->held = false; + } + }; + + $cache = new InMemoryCache(); + $clock = new TestClock(); + $task = static fn() => null; + $schedule = $this->schedule($cache, [$task])->lock($mutex); + + $generator = new TaskGenerator($schedule, $clock); + $this->assertNotNull($generator->getNextRunDate()); + + $clock->advance('+1 minute'); + $claimed = null; + + foreach ($generator->getTasks() as $context => $ignored) { + $claimed = $context; + } + + $this->assertNotNull($claimed); + + // The tick took the lock and gave it back; each of these takes it again on its own. + $taken = $mutex->acquired; + + $generator->markStarted($claimed); + $this->assertSame($taken + 1, $mutex->acquired); + + $generator->resolve($claimed, TaskStatus::Done); + $this->assertSame($taken + 2, $mutex->acquired); + $this->assertSame($mutex->acquired, $mutex->released, 'the lock is given back every time'); + } + + public function testAnOutcomeIsNotRecordedWhileAnotherProcessHoldsTheLock(): void + { + $mutex = new InMemoryMutex(); + $cache = new InMemoryCache(); + $clock = new TestClock(); + $task = static fn() => null; + $schedule = $this->schedule($cache, [$task])->lock($mutex); + + $generator = new TaskGenerator($schedule, $clock); + $this->assertNotNull($generator->getNextRunDate()); + + $clock->advance('+1 minute'); + $claimed = null; + + foreach ($generator->getTasks() as $context => $ignored) { + $claimed = $context; + } + + $this->assertNotNull($claimed); + + // Another process takes the schedule lock and keeps it. + $mutex->forceLock(); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('the schedule lock was held'); + + $generator->resolve($claimed, TaskStatus::Done); + } + + /** + * Claims the three runs a task missed while the schedule was down, without resolving any. + * + * @return list + */ + private function claimThreeMissedRuns( + ?InMemoryCache &$cache, + ?TestClock &$clock, + callable $task, + ?TaskGenerator &$generator, + ): array { + $cache = new InMemoryCache(); + $clock = new TestClock(); + $schedule = $this->schedule($cache, [$task]); + + (new Scheduler([$schedule], new CallableTaskHandler(), $clock))->tick(); + $clock->advance('+3 minutes'); + + $generator = new TaskGenerator($schedule, $clock); + $claimed = []; + + foreach ($generator->getTasks() as $context => $_) { + $claimed[] = $context; + } + + $this->assertCount(3, $claimed); + + return $claimed; + } + /** * Emits the due runs one by one and stops before starting the third, the way a process killed - * mid-tick leaves the two runs before it recorded and the ones behind it untouched. + * mid-tick leaves the two runs before it recorded, the one in hand claimed with nobody left to + * resolve it, and the ones behind it untouched. * * @param list $tasks */ @@ -314,22 +933,43 @@ private function crashAfterTwoRuns(InMemoryCache $cache, TestClock $clock, array foreach ($generator->getTasks() as $context => $task) { if (2 === $started) { + // The process dies here, holding a claim it will never resolve. break; } $handler->handle($task->getTask(), $context); + $generator->resolve($context, TaskStatus::Done); ++$started; } unset($generator); + + // Nothing is going to report what became of that claim, so the schedule takes it back + // once its reclaim window runs out. Everything below resumes past that point. + $clock->advance('+30 seconds'); } /** * @param list $tasks */ - private function tick(InMemoryCache $cache, TestClock $clock, array $tasks): void + private function tick(InMemoryCache $cache, TestClock $clock, array $tasks, int $reclaimAfter = 1): void { - (new Scheduler([$this->schedule($cache, $tasks)], new CallableTaskHandler(), $clock))->tick(); + $schedule = $this->schedule($cache, $tasks, $reclaimAfter); + + (new Scheduler([$schedule], new CallableTaskHandler(), $clock))->tick(); + } + + /** + * Leaves a claim of the given run in the checkpoint, as a process that never resolved it would. + */ + private function leaveClaim(InMemoryCache $cache, callable $task, DateTimeImmutable $at, string $owner): void + { + $cache->set($this->taskKey($task), [ + 'time' => $at->format('U.u'), + 'status' => 'claimed', + 'owner' => $owner, + 'at' => $at->format('U.u'), + ]); } /** @@ -343,12 +983,9 @@ private function taskKey(callable $task, string $scheduleName = 'default'): stri . hash('xxh3', RecurringTask::cron('* * * * *', $task)->getId()); } - /** - * @param list $tasks - */ - private function schedule(InMemoryCache $cache, array $tasks): Schedule + private function schedule(InMemoryCache $cache, array $tasks, int $reclaimAfter = 1): Schedule { - $schedule = (new Schedule())->stateful($cache); + $schedule = (new Schedule())->stateful($cache)->reclaimAfter($reclaimAfter); foreach ($tasks as $task) { $schedule->task(RecurringTask::cron('* * * * *', $task)); diff --git a/tests/SchedulerTest.php b/tests/SchedulerTest.php index 73549bd..64b0a2d 100644 --- a/tests/SchedulerTest.php +++ b/tests/SchedulerTest.php @@ -381,6 +381,98 @@ public function testFailingScheduleDoesNotKeepTheOtherSchedulesFromRunning(): vo $this->assertSame(['second'], $dispatcher->started); } + #[RequiresPhpExtension('pcntl')] + public function testEveryMissedRunOfACatchUpIsStartedWhileTheyQueueForOneSlot(): void + { + // Concurrent generation claims all three missed runs before starting any of them, and + // they share the one record. Mistaking the earlier two for runs some other scheduler + // took over loses the whole backlog but its last run, while the tick still reports + // three. + $clock = new TestClock(); + $cache = new InMemoryCache(); + + $schedule = (new Schedule())->stateful($cache)->task( + RecurringTask::every(60, static fn() => null)->withId('minute'), + ); + + $logger = new class extends AbstractLogger { + /** @var list */ + public array $warnings = []; + + public function log($level, string|Stringable $message, array $context = []): void + { + if (LogLevel::WARNING === $level) { + $this->warnings[] = (string) $message; + } + } + }; + + $scheduler = new Scheduler([$schedule], new CallableTaskHandler(), $clock, null, $logger); + $scheduler->tick(); + $clock->advance('+3 minutes'); + + $this->assertSame(3, $scheduler->tick(true, 1)); + + while ($scheduler->hasRunningChildren()) { + $scheduler->reapChildren(); + usleep(10_000); + } + + $this->assertSame([], $logger->warnings); + } + + #[RequiresPhpExtension('pcntl')] + public function testRunTakenOverWhileItWaitedForASlotIsNotStartedByThisScheduler(): void + { + $clock = new TestClock(); + $cache = new InMemoryCache(); + + $first = RecurringTask::cron('* * * * *', static fn() => null)->withId('first'); + $second = RecurringTask::cron('* * * * *', static fn() => null)->withId('second'); + $schedule = (new Schedule())->stateful($cache)->task($first, $second); + + $logger = new class extends AbstractLogger { + /** @var list */ + public array $warnings = []; + + public function log($level, string|Stringable $message, array $context = []): void + { + if (LogLevel::WARNING === $level) { + $this->warnings[] = (string) $message; + } + } + }; + + $scheduler = new Scheduler([$schedule], new CallableTaskHandler(), $clock, null, $logger); + $scheduler->tick(); + $clock->advance('+1 minute'); + + // Both runs are claimed as the tick is drained, and the second one waits behind the + // first. Another scheduler takes it over during that wait. + $taken = [ + 'time' => $clock->now()->format('U.u'), + 'status' => 'claimed', + 'owner' => 'another-host/1', + 'at' => $clock->now()->format('U.u'), + ]; + $key = 'yii_schedule_checkpoint_' . hash('xxh3', 'default') . '_' . hash('xxh3', $second->getId()); + $cache->replaceOnRead($key, $taken); + + $scheduler->tick(true); + + while ($scheduler->hasRunningChildren()) { + $scheduler->reapChildren(); + usleep(10_000); + } + + // Starting it here too would be the second time that run happened. + $this->assertCount(1, $logger->warnings); + $this->assertStringContainsString('taken over', $logger->warnings[0]); + + // And the scheduler that lost it does not get to say anything about it either. + $this->assertSame($taken, $cache->get($key)); + } + #[RequiresPhpExtension('pcntl')] public function testFailingScheduleDoesNotKeepTheOtherSchedulesFromRunningConcurrently(): void { @@ -594,7 +686,7 @@ public function testStandbyProcessDoesNotReplayRunsHandledByTheLockHolder(): voi $this->assertSame(5, $ran); } - public function testTaskFailureIsNotMaskedByACheckpointPersistenceFailure(): void + public function testRunIsNotHandedOutWhenItsClaimCannotBePersisted(): void { $cache = new InMemoryCache(); $clock = new TestClock(); @@ -613,8 +705,10 @@ public function testTaskFailureIsNotMaskedByACheckpointPersistenceFailure(): voi $clock->advance('+1 minute'); $cache->failWrites(); - $this->expectException(RuntimeException::class); - $this->expectExceptionMessage('Task failed.'); + // A run is claimed before it is handed out, so a checkpoint that cannot be written stops + // the task from running at all rather than letting it run unrecorded and be repeated. + $this->expectException(SchedulerRuntimeException::class); + $this->expectExceptionMessage('Unable to persist the checkpoint'); $scheduler->tick(); } diff --git a/tests/Support/InMemoryCache.php b/tests/Support/InMemoryCache.php index ea487a9..86c4711 100644 --- a/tests/Support/InMemoryCache.php +++ b/tests/Support/InMemoryCache.php @@ -7,6 +7,7 @@ use DateInterval; use Psr\SimpleCache\CacheInterface; +use function array_key_exists; use function sprintf; use function strlen; @@ -19,6 +20,8 @@ final class InMemoryCache implements CacheInterface private bool $writesFail = false; private int $writes = 0; private ?int $maximumKeyLength = null; + /** @var array */ + private array $replaceOnRead = []; /** * Makes every subsequent write report a failure, as a full or unavailable cache would. @@ -37,10 +40,27 @@ public function limitKeyLength(int $maximum): void $this->maximumKeyLength = $maximum; } + /** + * Hands the next read of this one key a value somebody else wrote, the way another process + * writing between two reads of it would. Reads through getMultiple() are left alone, so a + * checkpoint can be loaded without springing it. + */ + public function replaceOnRead(string $key, mixed $value): void + { + $this->replaceOnRead[$key] = $value; + } + public function get(string $key, mixed $default = null): mixed { $this->checkKey($key); + if (array_key_exists($key, $this->replaceOnRead)) { + /** @var mixed $replacement */ + $replacement = $this->replaceOnRead[$key]; + unset($this->replaceOnRead[$key]); + $this->values[$key] = $replacement; + } + return $this->values[$key] ?? $default; } @@ -85,7 +105,9 @@ public function clear(): bool public function getMultiple(iterable $keys, mixed $default = null): iterable { foreach ($keys as $key) { - yield $key => $this->get($key, $default); + $this->checkKey($key); + + yield $key => $this->values[$key] ?? $default; } }