From 3fe167d4cb145ca2d213b36777b06b8e561ceb50 Mon Sep 17 00:00:00 2001
From: KalimeroMK <zbogoevski@gmail.com>
Date: Sat, 31 Oct 2026 17:58:24 +0100
Subject: [PATCH 1/7] Record what became of a run, not only that it was emitted

A record said a run had been emitted and nothing more, so the scheduler could
not tell a run it had made from one it had merely handed out. Draining the
generator claimed every due run before any of them started, and a daemon told
to stop then left behind runs recorded as emitted that never ran: "Stopping,
{count} due task(s) of this tick are not started."

A run is now claimed as it is handed out and recorded as done, failed, skipped
or cancelled once its outcome is known. The runs a stopping daemon never gets
to are given back to the schedule and stay due.

The runs of a task are claimed in order but resolved as their outcomes arrive,
so the record moves one way only: a resolution of an earlier run than the one
on record is dropped, and a run given back takes the record no further forward
than it already is. Without that, a later run resolved before an earlier one
dragged the record back over the ones in between and handed them out again.

A claim carries the process that made it and the moment it was made. A process
of this host that is gone is noticed at once with ext-posix; anywhere else the
claim is taken back after Schedule::reclaimAfter(), a day by default. Until
then the run is left alone, because handing out a run that is still going
would run the task a second time. A claim running out of window changes where
its task resumes without changing anything that was stored, so the pending
runs a long-lived scheduler is holding are worked out again when it does.

A claim is written before the run is handed out, so a checkpoint that cannot
be written now stops the run from being made at all rather than letting it run
unrecorded and be repeated.
---
 CHANGELOG.md                          |   1 +
 README.md                             |  23 ++
 src/Generator/Checkpoint.php          | 328 +++++++++++++++++++++-----
 src/Generator/TaskGenerator.php       |  65 +++--
 src/Schedule.php                      |  39 +++
 src/Scheduler.php                     |  71 +++++-
 src/TaskStatus.php                    |  33 +++
 tests/Generator/TaskGeneratorTest.php | 296 +++++++++++++++++++++--
 tests/SchedulerTest.php               |   8 +-
 9 files changed, 748 insertions(+), 116 deletions(-)
 create mode 100644 src/TaskStatus.php

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3d5bc3b..094a8d4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 ## 1.0.0 under development
 
+- 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..64e547d 100644
--- a/README.md
+++ b/README.md
@@ -208,6 +208,29 @@ $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. One made by a process of this host is
+noticed as soon as that process is gone, which needs `ext-posix`; one made anywhere else 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 the whole of a run, counted from the moment it is claimed rather than
+from the moment it starts: under `--max-processes` a due run waits for a free slot, and that
+wait is inside the window too. Taking a run back while its process still means to make it runs
+the task a second time.
+
+The window only applies where the process cannot be asked about. A claim of this host is taken
+back as soon as its process is gone, whatever the window says.
+
 ## 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..38352de 100644
--- a/src/Generator/Checkpoint.php
+++ b/src/Generator/Checkpoint.php
@@ -4,69 +4,88 @@
 
 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. One whose process is gone is handed out again; a
+ * process of another host, or of a host without ext-posix, cannot be asked about, so there the
+ * claim stands until the reclaim window runs out — handing out a run that is still going would
+ * run the task twice over.
  *
  * @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;
+
     private ?DateTimeImmutable $time = null;
-    /** @var array<string, DateTimeImmutable> Task id to the time of its last emitted run. */
-    private array $lastRuns = [];
+    /** @var array<string, Record> Run, outcome, and the owner and moment of a claim. */
+    private array $records = [];
     private bool $locked = false;
     private int $revision = 0;
 
-    /**
-     * @param list<string> $taskIds Ids of the tasks of the schedule, whose records are read back.
-     */
+    /** @var list<string> 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.
-     *
-     * 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.
+     * 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.
      */
-    public function acquire(): bool
+    /**
+     * @param list<string> $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(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 +97,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 +112,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 +166,12 @@ 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.
+     * 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 +197,74 @@ 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 nobody is left to hear from.
+     */
+    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. A resolution of an earlier run than the one on record
+     * is dropped: the runs of a task are claimed in order but resolved as their outcomes arrive,
+     * and moving the record back would hand out the ones in between a second time.
+     */
+    public function resolve(string $taskId, DateTimeImmutable $time, TaskStatus $status): void
+    {
+        $record = $this->records[$taskId] ?? null;
+
+        if (null !== $record && $time < $record[0]) {
+            return;
+        }
+
+        $this->write($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
+    {
+        $record = $this->records[$taskId] ?? null;
+
+        if (null !== $record && $time > $record[0]) {
+            return;
+        }
+
+        $this->write($taskId, $time, TaskStatus::Released);
+    }
+
     private function writePosition(DateTimeImmutable $now): void
     {
         $this->time = $now;
@@ -183,6 +280,57 @@ 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: a run handed out twice 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 +351,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 +378,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 +390,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 +461,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..fff421c 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,17 @@ 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.
+                $this->checkpoint->write($task->getId(), $time, TaskStatus::Claimed, at: $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 +111,7 @@ public function getTasks(): iterable
      */
     public function getNextRunDate(): ?DateTimeImmutable
     {
-        if (!$this->checkpoint->acquire()) {
+        if (!$this->checkpoint->acquire($this->taskIds())) {
             return null;
         }
 
@@ -129,11 +124,41 @@ 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);
+    }
+
+    /**
+     * 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);
+    }
+
+    /**
+     * @return list<string>
+     */
+    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 +171,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..1f12113 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<callable(TaskContext, RecurringTask): void> */
     private array $beforeListeners = [];
     /** @var list<callable(TaskContext, RecurringTask, mixed): void> */
@@ -89,6 +94,40 @@ 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. A
+     * gone process of this host is noticed at once with ext-posix; this covers the rest.
+     *
+     * It has to outlast the whole of a run, from being claimed to being over — which includes
+     * the wait for a free slot under `--max-processes`, not only the work itself. 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..1752ec1 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,17 +192,23 @@ 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;
                 }
 
@@ -442,13 +458,26 @@ 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);
     }
 
     /**
@@ -477,7 +506,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 +528,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 +554,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 +580,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 +730,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 +741,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 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Scheduler;
+
+/**
+ * What became of a run of a task, as the schedule checkpoint records it.
+ */
+enum TaskStatus: string
+{
+    /**
+     * Handed out, outcome not known: a run in flight, or one left by a process that is gone. The
+     * first is left alone; the second is handed out again, at once when the process can be shown
+     * to be gone and otherwise once the reclaim window of the schedule runs out.
+     */
+    case Claimed = 'claimed';
+
+    /** The run was made and the task reported no error. */
+    case Done = 'done';
+
+    /** The run was made and the task failed. */
+    case Failed = 'failed';
+
+    /** Not made: a previous run was still going, or a newer missed run superseded it. */
+    case Skipped = 'skipped';
+
+    /** Claimed and handed back unstarted, so the run is due again. */
+    case Released = 'released';
+
+    /** The run was not made: a listener cancelled it. */
+    case Cancelled = 'cancelled';
+}
diff --git a/tests/Generator/TaskGeneratorTest.php b/tests/Generator/TaskGeneratorTest.php
index 3f70069..2caa05a 100644
--- a/tests/Generator/TaskGeneratorTest.php
+++ b/tests/Generator/TaskGeneratorTest.php
@@ -5,7 +5,9 @@
 namespace Yiisoft\Scheduler\Tests\Generator;
 
 use ArrayObject;
+use DateTimeImmutable;
 use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\Attributes\RequiresPhpExtension;
 use PHPUnit\Framework\TestCase;
 use Yiisoft\Scheduler\Exception\RuntimeException;
 use Yiisoft\Scheduler\Generator\TaskGenerator;
@@ -14,9 +16,14 @@
 use Yiisoft\Scheduler\Schedule;
 use Yiisoft\Scheduler\Scheduler;
 use Yiisoft\Scheduler\Tests\Support\InMemoryCache;
+use Yiisoft\Scheduler\TaskContext;
+use Yiisoft\Scheduler\TaskStatus;
 use Yiisoft\Scheduler\Tests\Support\TestClock;
 
+use function gethostname;
 use function hash;
+use function pcntl_fork;
+use function pcntl_waitpid;
 
 /**
  * The checkpoint records which runs of a shared due time were already emitted, so that a process
@@ -106,6 +113,199 @@ public function testRunsEmittedBeforeACrashAreNotRepeatedAfterATaskIsAdded(): vo
         $this->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 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());
+    }
+
     public function testRunsAlreadyRecordedAreNotWrittenToTheCheckpointAgain(): void
     {
         $log = new ArrayObject();
@@ -127,10 +327,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 +395,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 +412,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 +442,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 +503,40 @@ public function testMalformedCheckpointIsRejected(mixed $state): void
         $this->tick($cache, new TestClock(), [$task]);
     }
 
+    /**
+     * Claims the three runs a task missed while the schedule was down, without resolving any.
+     *
+     * @return list<TaskContext>
+     */
+    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<callable> $tasks
      */
@@ -314,22 +553,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<callable> $tasks
      */
-    private function tick(InMemoryCache $cache, TestClock $clock, array $tasks): void
+    private function tick(InMemoryCache $cache, TestClock $clock, array $tasks, int $reclaimAfter = 1): void
+    {
+        $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
     {
-        (new Scheduler([$this->schedule($cache, $tasks)], new CallableTaskHandler(), $clock))->tick();
+        $cache->set($this->taskKey($task), [
+            'time' => $at->format('U.u'),
+            'status' => 'claimed',
+            'owner' => $owner,
+            'at' => $at->format('U.u'),
+        ]);
     }
 
     /**
@@ -346,9 +606,9 @@ private function taskKey(callable $task, string $scheduleName = 'default'): stri
     /**
      * @param list<callable> $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..c83623d 100644
--- a/tests/SchedulerTest.php
+++ b/tests/SchedulerTest.php
@@ -594,7 +594,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 +613,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();
     }

From bc7b6138e7938db1f9fcf465d1bea09eb45c0886 Mon Sep 17 00:00:00 2001
From: KalimeroMK <zbogoevski@gmail.com>
Date: Sun, 1 Nov 2026 00:45:54 +0100
Subject: [PATCH 2/7] Start the reclaim window when a run starts, not when it
 was claimed

A run is claimed as it is handed out, which is not when it begins. Under
`--max-processes` it then waits for a free slot, and in a tick handing out
several runs the ones behind a slow task wait for it to finish. Both waits
were charged against the window deciding when another process may take the
run over, so a long enough wait had the run handed out a second time while
the process holding it was still about to make it.

The claim is now stamped as it is made rather than as the tick began, and
`markStarted()` restamps it once the run actually starts.
---
 CHANGELOG.md                          |   1 +
 README.md                             |   7 +-
 src/Generator/Checkpoint.php          |  30 ++++++-
 src/Generator/TaskGenerator.php       |  15 +++-
 src/Schedule.php                      |   6 +-
 src/Scheduler.php                     |   9 +++
 tests/Generator/TaskGeneratorTest.php | 109 ++++++++++++++++++++++++++
 7 files changed, 167 insertions(+), 10 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 094a8d4..72c980f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 ## 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)
diff --git a/README.md b/README.md
index 64e547d..8329251 100644
--- a/README.md
+++ b/README.md
@@ -223,10 +223,9 @@ taken back after a window, set with `reclaimAfter()` and a day by default:
 $schedule->reclaimAfter(900); // or a DateInterval
 ```
 
-The window has to outlast the whole of a run, counted from the moment it is claimed rather than
-from the moment it starts: under `--max-processes` a due run waits for a free slot, and that
-wait is inside the window too. Taking a run back while its process still means to make it runs
-the task a second time.
+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.
 
 The window only applies where the process cannot be asked about. A claim of this host is taken
 back as soon as its process is gone, whatever the window says.
diff --git a/src/Generator/Checkpoint.php b/src/Generator/Checkpoint.php
index 38352de..6f3b169 100644
--- a/src/Generator/Checkpoint.php
+++ b/src/Generator/Checkpoint.php
@@ -69,8 +69,7 @@ public function __construct(
     /**
      * 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.
-     */
-    /**
+     *
      * @param list<string> $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.
      */
@@ -165,6 +164,33 @@ public function write(
         }
     }
 
+    /**
+     * 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.
+     */
+    public function markStarted(string $taskId, DateTimeImmutable $time, DateTimeImmutable $at): void
+    {
+        $record = $this->records[$taskId] ?? null;
+
+        if (null === $record) {
+            return;
+        }
+
+        [$recordedTime, $status, $owner] = $record;
+
+        if (TaskStatus::Claimed !== $status || $time != $recordedTime || self::owner() !== $owner) {
+            return;
+        }
+
+        $this->write($taskId, $time, TaskStatus::Claimed, $owner, $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.
diff --git a/src/Generator/TaskGenerator.php b/src/Generator/TaskGenerator.php
index fff421c..c097a21 100644
--- a/src/Generator/TaskGenerator.php
+++ b/src/Generator/TaskGenerator.php
@@ -89,7 +89,11 @@ public function getTasks(): iterable
 
                 // 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.
-                $this->checkpoint->write($task->getId(), $time, TaskStatus::Claimed, at: $now);
+                //
+                // 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());
 
                 yield $context => $task;
             }
@@ -130,6 +134,15 @@ 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.
+     */
+    public function markStarted(TaskContext $context): void
+    {
+        $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.
diff --git a/src/Schedule.php b/src/Schedule.php
index 1f12113..7900c02 100644
--- a/src/Schedule.php
+++ b/src/Schedule.php
@@ -98,9 +98,9 @@ public function stateful(CacheInterface $cache): static
      * Sets how long a claimed run is left to its process before the schedule takes it back. A
      * gone process of this host is noticed at once with ext-posix; this covers the rest.
      *
-     * It has to outlast the whole of a run, from being claimed to being over — which includes
-     * the wait for a free slot under `--max-processes`, not only the work itself. Taking a run
-     * back while its process still means to make it runs the task twice.
+     * 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.
      */
diff --git a/src/Scheduler.php b/src/Scheduler.php
index 1752ec1..9cb248f 100644
--- a/src/Scheduler.php
+++ b/src/Scheduler.php
@@ -212,6 +212,10 @@ public function tick(bool $concurrent = false, ?int $maxProcesses = null): int
                     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.
+                $this->markStarted($context);
+
                 $this->forkTask($task, $context);
             } catch (Throwable $error) {
                 $failure = $this->firstFailure($failure, $error);
@@ -480,6 +484,11 @@ private function release(TaskContext $context): void
         $this->generators[$context->scheduleName]->release($context);
     }
 
+    private function markStarted(TaskContext $context): void
+    {
+        $this->generators[$context->scheduleName]->markStarted($context);
+    }
+
     /**
      * Dispatches the pre-run event and the schedule's before listeners.
      *
diff --git a/tests/Generator/TaskGeneratorTest.php b/tests/Generator/TaskGeneratorTest.php
index 2caa05a..4e75afb 100644
--- a/tests/Generator/TaskGeneratorTest.php
+++ b/tests/Generator/TaskGeneratorTest.php
@@ -503,6 +503,115 @@ 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']);
+    }
+
     /**
      * Claims the three runs a task missed while the schedule was down, without resolving any.
      *

From 8c12fcb2695987bdad69255a6b15561af192fa66 Mon Sep 17 00:00:00 2001
From: KalimeroMK <zbogoevski@gmail.com>
Date: Mon, 2 Nov 2026 00:52:16 +0100
Subject: [PATCH 3/7] Judge a claim by the state as it is, not as it was read

An outcome is reported long after the claim was made, and the process reporting
it had not read the state since. Both guards it passed through were therefore
guards against its own stale copy.

Two schedulers, one cache. A takes the run due at 00:01 and is still running it;
B takes the run due at 00:02 and finishes it. A then reaps its child and records
00:01 as done, because the record A last read says 00:01 is A's. The record moves
behind a run that is over, and 00:02 is handed out again.

The same copy is what made a claim safe to start. A run that waited for a free
process slot for longer than the reclaim window has been taken over by then, and
the scheduler holding it started it anyway: the window was reset after the wait,
not checked before the run.

Every write about a claimed run now re-reads that record first and goes ahead
only while the state still holds this process's own claim of it. markStarted()
says so too, and a run that is no longer this scheduler's is logged and left to
whoever has it rather than started a second time.

The record of a task stands for the newest run claimed, so a catch-up that
claimed several still resolves only that one and still hands back from the
earliest, as before.
---
 src/Generator/Checkpoint.php          | 89 ++++++++++++++++++++++-----
 src/Generator/TaskGenerator.php       |  6 +-
 src/Scheduler.php                     | 17 ++++-
 tests/Generator/TaskGeneratorTest.php | 83 +++++++++++++++++++++++++
 tests/SchedulerTest.php               | 52 ++++++++++++++++
 tests/Support/InMemoryCache.php       | 24 +++++++-
 6 files changed, 248 insertions(+), 23 deletions(-)

diff --git a/src/Generator/Checkpoint.php b/src/Generator/Checkpoint.php
index 6f3b169..454da6a 100644
--- a/src/Generator/Checkpoint.php
+++ b/src/Generator/Checkpoint.php
@@ -173,22 +173,22 @@ public function write(
      *
      * 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): void
+    public function markStarted(string $taskId, DateTimeImmutable $time, DateTimeImmutable $at): bool
     {
-        $record = $this->records[$taskId] ?? null;
+        $claim = $this->ourClaim($taskId);
 
-        if (null === $record) {
-            return;
+        if (null === $claim || $time != $claim[0]) {
+            return false;
         }
 
-        [$recordedTime, $status, $owner] = $record;
+        $this->write($taskId, $time, TaskStatus::Claimed, self::owner(), $at);
 
-        if (TaskStatus::Claimed !== $status || $time != $recordedTime || self::owner() !== $owner) {
-            return;
-        }
-
-        $this->write($taskId, $time, TaskStatus::Claimed, $owner, $at);
+        return true;
     }
 
     /**
@@ -261,15 +261,20 @@ public function hasReclaimableClaim(DateTimeImmutable $now, DateInterval $reclai
     }
 
     /**
-     * Records what became of a claimed run. A resolution of an earlier run than the one on record
-     * is dropped: the runs of a task are claimed in order but resolved as their outcomes arrive,
-     * and moving the record back would hand out the ones in between a second time.
+     * 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
     {
-        $record = $this->records[$taskId] ?? null;
+        $claim = $this->ourClaim($taskId);
 
-        if (null !== $record && $time < $record[0]) {
+        // 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;
         }
 
@@ -282,15 +287,65 @@ public function resolve(string $taskId, DateTimeImmutable $time, TaskStatus $sta
      */
     public function giveBack(string $taskId, DateTimeImmutable $time): void
     {
-        $record = $this->records[$taskId] ?? null;
+        $claim = $this->ourClaim($taskId);
 
-        if (null !== $record && $time > $record[0]) {
+        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;
diff --git a/src/Generator/TaskGenerator.php b/src/Generator/TaskGenerator.php
index c097a21..1d30e00 100644
--- a/src/Generator/TaskGenerator.php
+++ b/src/Generator/TaskGenerator.php
@@ -137,10 +137,12 @@ public function resolve(TaskContext $context, TaskStatus $status): void
     /**
      * 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): void
+    public function markStarted(TaskContext $context): bool
     {
-        $this->checkpoint->markStarted($context->taskId, $context->triggeredAt, $this->clock->now());
+        return $this->checkpoint->markStarted($context->taskId, $context->triggeredAt, $this->clock->now());
     }
 
     /**
diff --git a/src/Scheduler.php b/src/Scheduler.php
index 9cb248f..a4158c2 100644
--- a/src/Scheduler.php
+++ b/src/Scheduler.php
@@ -214,7 +214,17 @@ public function tick(bool $concurrent = false, ?int $maxProcesses = null): int
 
                 // 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.
-                $this->markStarted($context);
+                // 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) {
@@ -484,9 +494,10 @@ private function release(TaskContext $context): void
         $this->generators[$context->scheduleName]->release($context);
     }
 
-    private function markStarted(TaskContext $context): void
+    /** Whether the run is still this process's to start. */
+    private function markStarted(TaskContext $context): bool
     {
-        $this->generators[$context->scheduleName]->markStarted($context);
+        return $this->generators[$context->scheduleName]->markStarted($context);
     }
 
     /**
diff --git a/tests/Generator/TaskGeneratorTest.php b/tests/Generator/TaskGeneratorTest.php
index 4e75afb..b138f8d 100644
--- a/tests/Generator/TaskGeneratorTest.php
+++ b/tests/Generator/TaskGeneratorTest.php
@@ -215,6 +215,89 @@ public function testGivingBackSeveralRunsLeavesACompletedOneAlone(): void
         $this->assertSame(['task', 'task'], $log->getArrayCopy());
     }
 
+    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 testClaimExpiringUnderALongLivedSchedulerIsStillHandedOut(): void
     {
         $log = new ArrayObject();
diff --git a/tests/SchedulerTest.php b/tests/SchedulerTest.php
index c83623d..950158e 100644
--- a/tests/SchedulerTest.php
+++ b/tests/SchedulerTest.php
@@ -381,6 +381,58 @@ public function testFailingScheduleDoesNotKeepTheOtherSchedulesFromRunning(): vo
         $this->assertSame(['second'], $dispatcher->started);
     }
 
+    #[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<string> */
+            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
     {
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<string, mixed> */
+    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;
         }
     }
 

From 7628ee4e045b388a798b2006839e7647f58f22c3 Mon Sep 17 00:00:00 2001
From: KalimeroMK <zbogoevski@gmail.com>
Date: Mon, 2 Nov 2026 00:52:41 +0100
Subject: [PATCH 4/7] Put a run handed back in front of the scheduler that
 handed it back

A graceful stop hands back the runs it never started, and the record then says
the task resumes before the earliest of them. The pending runs the scheduler is
holding say otherwise: they were worked out while those runs were claimed, and
the only occurrence of the task left in them is the one after.

A scheduler that is stopped and used again therefore reported the next run as
due and skipped the one it had just handed back, while a process starting fresh
on the same state ran it. Advancing that scheduler replaced the handed-back run
with the new one for good.

Handing a run back now drops the pending runs, so the next question about them
is answered from the record.
---
 src/Generator/TaskGenerator.php       |  6 ++++++
 tests/Generator/TaskGeneratorTest.php | 31 +++++++++++++++++++++++++++
 2 files changed, 37 insertions(+)

diff --git a/src/Generator/TaskGenerator.php b/src/Generator/TaskGenerator.php
index 1d30e00..1a6141f 100644
--- a/src/Generator/TaskGenerator.php
+++ b/src/Generator/TaskGenerator.php
@@ -152,6 +152,12 @@ public function markStarted(TaskContext $context): bool
     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;
     }
 
     /**
diff --git a/tests/Generator/TaskGeneratorTest.php b/tests/Generator/TaskGeneratorTest.php
index b138f8d..7a53704 100644
--- a/tests/Generator/TaskGeneratorTest.php
+++ b/tests/Generator/TaskGeneratorTest.php
@@ -298,6 +298,37 @@ public function testLateOutcomeDoesNotMoveTheRecordBehindARunAnotherSchedulerFin
         $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();

From 347cb8b63d6a193269b97bf2183fccc273c1782a Mon Sep 17 00:00:00 2001
From: KalimeroMK <zbogoevski@gmail.com>
Date: Mon, 2 Nov 2026 13:15:56 +0100
Subject: [PATCH 5/7] Start every run of a catch-up, and record an outcome
 under the schedule lock

Two things the last review found, both of them mine.

The guard that keeps a run from starting after another scheduler took it over was comparing the
run against the record exactly. 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 task that
missed 00:01, 00:02 and 00:03 therefore had the first two refused as takeovers by the very
scheduler that had claimed them. Under `tick(true, 1)` only 00:03 ran, while the tick reported
three. A run at or before the record is this process's own and starts; only the newest moves the
window, so an earlier one no longer drags the record back to its own time.

Rereading the record before writing narrowed the window where two schedulers disagree about a
task, but did not close it: between that read and the write, another scheduler can finish a
later run and the write then puts the record back where that run started. The read, the decision
and the write now happen with the schedule lock held, as the claim itself does. A tick already
holds that lock while it generates and a sequential run resolves inside it, so the lock is taken
only when this process is not already holding it; a mutex is not reentrant and taking it twice
would wait on this process. Unlike the tick, which moves on when another process has the
schedule, this waits: what became of a run that already happened has to be recorded. A lock that
cannot be had in ten seconds is reported rather than written through.
---
 src/Generator/Checkpoint.php          | 112 +++++++++++++++++++---
 tests/Generator/TaskGeneratorTest.php | 130 +++++++++++++++++++++++++-
 tests/SchedulerTest.php               |  40 ++++++++
 3 files changed, 264 insertions(+), 18 deletions(-)

diff --git a/src/Generator/Checkpoint.php b/src/Generator/Checkpoint.php
index 454da6a..566bc4a 100644
--- a/src/Generator/Checkpoint.php
+++ b/src/Generator/Checkpoint.php
@@ -51,6 +51,13 @@ final class 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<string, Record> Run, outcome, and the owner and moment of a claim. */
     private array $records = [];
@@ -180,15 +187,9 @@ public function write(
      */
     public function markStarted(string $taskId, DateTimeImmutable $time, DateTimeImmutable $at): bool
     {
-        $claim = $this->ourClaim($taskId);
-
-        if (null === $claim || $time != $claim[0]) {
-            return false;
-        }
-
-        $this->write($taskId, $time, TaskStatus::Claimed, self::owner(), $at);
-
-        return true;
+        return $this->exclusively(function () use ($taskId, $time, $at): bool {
+            return $this->startUnderLock($taskId, $time, $at);
+        });
     }
 
     /**
@@ -269,6 +270,93 @@ public function hasReclaimableClaim(DateTimeImmutable $now, DateInterval $reclai
      * 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);
 
@@ -281,11 +369,7 @@ public function resolve(string $taskId, DateTimeImmutable $time, TaskStatus $sta
         $this->write($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
+    private function giveBackUnderLock(string $taskId, DateTimeImmutable $time): void
     {
         $claim = $this->ourClaim($taskId);
 
diff --git a/tests/Generator/TaskGeneratorTest.php b/tests/Generator/TaskGeneratorTest.php
index 7a53704..9581ca1 100644
--- a/tests/Generator/TaskGeneratorTest.php
+++ b/tests/Generator/TaskGeneratorTest.php
@@ -9,6 +9,7 @@
 use PHPUnit\Framework\Attributes\DataProvider;
 use PHPUnit\Framework\Attributes\RequiresPhpExtension;
 use PHPUnit\Framework\TestCase;
+use Yiisoft\Mutex\MutexInterface;
 use Yiisoft\Scheduler\Exception\RuntimeException;
 use Yiisoft\Scheduler\Generator\TaskGenerator;
 use Yiisoft\Scheduler\Handler\CallableTaskHandler;
@@ -16,6 +17,7 @@
 use Yiisoft\Scheduler\Schedule;
 use Yiisoft\Scheduler\Scheduler;
 use Yiisoft\Scheduler\Tests\Support\InMemoryCache;
+use Yiisoft\Scheduler\Tests\Support\InMemoryMutex;
 use Yiisoft\Scheduler\TaskContext;
 use Yiisoft\Scheduler\TaskStatus;
 use Yiisoft\Scheduler\Tests\Support\TestClock;
@@ -24,7 +26,6 @@
 use function hash;
 use function pcntl_fork;
 use function pcntl_waitpid;
-
 /**
  * The checkpoint records which runs of a shared due time were already emitted, so that a process
  * resuming after a crash replays the rest and only the rest.
@@ -33,6 +34,8 @@
  * moves the position of every task after it, while the runs recorded before the crash stay as
  * they were.
  */
+use function sprintf;
+
 final class TaskGeneratorTest extends TestCase
 {
     public function testRunsEmittedBeforeACrashAreNotRepeated(): void
@@ -215,6 +218,40 @@ public function testGivingBackSeveralRunsLeavesACompletedOneAlone(): void
         $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();
@@ -726,6 +763,94 @@ public function testMarkingAResolvedRunAsStartedLeavesItsRecordAlone(): void
         $this->assertSame('done', $record['status']);
     }
 
+    /**
+     * @param list<callable> $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.
      *
@@ -826,9 +951,6 @@ private function taskKey(callable $task, string $scheduleName = 'default'): stri
             . hash('xxh3', RecurringTask::cron('* * * * *', $task)->getId());
     }
 
-    /**
-     * @param list<callable> $tasks
-     */
     private function schedule(InMemoryCache $cache, array $tasks, int $reclaimAfter = 1): Schedule
     {
         $schedule = (new Schedule())->stateful($cache)->reclaimAfter($reclaimAfter);
diff --git a/tests/SchedulerTest.php b/tests/SchedulerTest.php
index 950158e..64b0a2d 100644
--- a/tests/SchedulerTest.php
+++ b/tests/SchedulerTest.php
@@ -381,6 +381,46 @@ 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<string> */
+            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
     {

From 27946aad1ceba95d30e878d895d1af09a8901a32 Mon Sep 17 00:00:00 2001
From: "github-actions[bot]" <github-actions[bot]@users.noreply.github.com>
Date: Mon, 2 Nov 2026 12:16:44 +0000
Subject: [PATCH 6/7] Apply PHP CS Fixer and Rector changes (CI)

---
 src/Generator/Checkpoint.php | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/src/Generator/Checkpoint.php b/src/Generator/Checkpoint.php
index 566bc4a..bb657c0 100644
--- a/src/Generator/Checkpoint.php
+++ b/src/Generator/Checkpoint.php
@@ -187,9 +187,7 @@ public function write(
      */
     public function markStarted(string $taskId, DateTimeImmutable $time, DateTimeImmutable $at): bool
     {
-        return $this->exclusively(function () use ($taskId, $time, $at): bool {
-            return $this->startUnderLock($taskId, $time, $at);
-        });
+        return $this->exclusively(fn(): bool => $this->startUnderLock($taskId, $time, $at));
     }
 
     /**

From 767a13638f808fa79f52edea2c6fe46664b3d59e Mon Sep 17 00:00:00 2001
From: KalimeroMK <zbogoevski@gmail.com>
Date: Mon, 2 Nov 2026 15:16:27 +0100
Subject: [PATCH 7/7] Say what the reclaim window does: every claim, live
 process or not

The window was documented as applying only where the owner's process cannot
be asked about, while it has always applied to every claim. Asking about the
process only brings the window forward: a live pid is no promise that the run
is still going, since a pid is given to something else once its process is
gone, and a claim waiting on one of those would stand forever.
---
 README.md                             | 13 ++++++-----
 src/Generator/Checkpoint.php          | 15 ++++++++-----
 src/Schedule.php                      |  5 +++--
 tests/Generator/TaskGeneratorTest.php | 32 +++++++++++++++++++++++++++
 4 files changed, 52 insertions(+), 13 deletions(-)

diff --git a/README.md b/README.md
index 8329251..fca811f 100644
--- a/README.md
+++ b/README.md
@@ -215,9 +215,8 @@ A run is claimed in the checkpoint as it is handed out and recorded as `done`, `
 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. One made by a process of this host is
-noticed as soon as that process is gone, which needs `ext-posix`; one made anywhere else is
-taken back after a window, set with `reclaimAfter()` and a day by default:
+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
@@ -227,8 +226,12 @@ The window has to outlast a run, counted from the moment it starts. Under `--max
 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.
 
-The window only applies where the process cannot be asked about. A claim of this host is taken
-back as soon as its process is gone, whatever the window says.
+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
 
diff --git a/src/Generator/Checkpoint.php b/src/Generator/Checkpoint.php
index bb657c0..c20e030 100644
--- a/src/Generator/Checkpoint.php
+++ b/src/Generator/Checkpoint.php
@@ -35,10 +35,11 @@
  * 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. One whose process is gone is handed out again; a
- * process of another host, or of a host without ext-posix, cannot be asked about, so there the
- * claim stands until the reclaim window runs out — handing out a run that is still going would
- * run the task twice over.
+ * 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
  *
@@ -224,7 +225,8 @@ public function markInitialPosition(DateTimeImmutable $now): void
 
     /**
      * 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 nobody is left to hear from.
+     * 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
     {
@@ -466,7 +468,8 @@ private static function owner(): string
 
     /**
      * Only a process of this host can be asked about, and only with ext-posix. Anything else
-     * reads as still there: a run handed out twice is worse than one picked up late.
+     * 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
     {
diff --git a/src/Schedule.php b/src/Schedule.php
index 7900c02..68bf420 100644
--- a/src/Schedule.php
+++ b/src/Schedule.php
@@ -95,8 +95,9 @@ public function stateful(CacheInterface $cache): static
     }
 
     /**
-     * Sets how long a claimed run is left to its process before the schedule takes it back. A
-     * gone process of this host is noticed at once with ext-posix; this covers the rest.
+     * 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
diff --git a/tests/Generator/TaskGeneratorTest.php b/tests/Generator/TaskGeneratorTest.php
index 9581ca1..adb401b 100644
--- a/tests/Generator/TaskGeneratorTest.php
+++ b/tests/Generator/TaskGeneratorTest.php
@@ -457,6 +457,38 @@ public function testClaimOfAProcessThatIsStillThereIsLeftAlone(): void
         $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();