Repository navigation
Record what became of a run, not only that it was emitted - #19
KalimeroMK wants to merge 7 commits into
Conversation
Reviewer's GuideThe scheduler now persists an owned claim before handing out each run, records its final outcome after execution, releases runs prevented by shutdown, and safely reclaims unresolved claims based on same-host process liveness or a configurable timeout. Sequence diagram for claiming and resolving a scheduled runsequenceDiagram
participant Scheduler
participant TaskGenerator
participant Checkpoint
participant Task
Scheduler->>TaskGenerator: getTasks()
TaskGenerator->>Checkpoint: write(taskId, time, Claimed, owner, at)
Checkpoint-->>TaskGenerator: claim persisted
TaskGenerator-->>Scheduler: yield context, task
Scheduler->>Task: runTask(task, context)
Task-->>Scheduler: TaskStatus
Scheduler->>TaskGenerator: resolve(context, status)
TaskGenerator->>Checkpoint: write(taskId, triggeredAt, status)
State diagram for a persisted task runstateDiagram-v2
[*] --> Claimed: write(..., TaskStatus.Claimed)
Claimed --> Done: resolve(..., TaskStatus.Done)
Claimed --> Failed: resolve(..., TaskStatus.Failed)
Claimed --> Skipped: resolve(..., TaskStatus.Skipped)
Claimed --> Cancelled: resolve(..., TaskStatus.Cancelled)
Claimed --> Due: release(context)
Claimed --> Due: isAbandoned(...)
Due --> Claimed: handed out again
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 5 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="src/Generator/Checkpoint.php" line_range="165" />
<code_context>
+ $owner = $claimed ? $owner ?? self::owner() : null;
+ $at = $claimed ? $at ?? $time : null;
+
+ $this->records[$taskId] = [$time, $status, $owner, $at];
if (null === $this->cache) {
</code_context>
<issue_to_address>
**Completed runs become due again**
When concurrent catch-up emits multiple due occurrences for one task and they resolve or release out of order, `Checkpoint::write()` replaces the record with an older timestamp, while `TaskGenerator::resolve()` and `release()` pop saved claims by task ID rather than matching the context. The checkpoint moves behind a completed or skipped run, so a later tick can emit it again.
Match saved claims to their contexts and prevent an older resolution or release from replacing a newer checkpoint.
Also at `src/Generator/TaskGenerator.php:154-180`, `src/Scheduler.php:210-212`, `src/Scheduler.php:565`.
</issue_to_address>
### Comment 2
<location path="src/Schedule.php" line_range="112-122" />
<code_context>
+ *
+ * @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)) {
+ if ($after <= 0) {
+ throw new Exception\LogicException('The reclaim window must be greater than zero.');
+ }
+
+ $after = new DateInterval(sprintf('PT%dS', $after));
+ }
+
+ $this->reclaimAfter = $after;
+
+ return $this;
</code_context>
<issue_to_address>
**Live runs are reclaimed immediately**
When a caller passes a zero or negative `DateInterval` to `reclaimAfter()`, `Schedule::reclaimAfter()` accepts the interval, so `Checkpoint::isAbandoned()` treats a live claim as expired immediately or prematurely. Another scheduler can hand out the run while its original task is executing.
Reject zero or negative `DateInterval` values in `reclaimAfter()`.
</issue_to_address>
### Comment 3
<location path="src/Generator/TaskGenerator.php" line_range="206-212" />
<code_context>
+ if (null !== $record) {
+ // An abandoned claim stands for a run that reported nothing and never will, so
+ // the task goes back to just before it and the run is handed out again.
+ $abandoned = $this->checkpoint->isAbandoned(
+ $task->getId(),
+ $this->clock->now(),
+ $this->schedule->getReclaimAfter(),
+ );
+
+ $seedTime = $abandoned ? $record[0]->modify('-1 microsecond') : $record[0];
+ }
+
</code_context>
<issue_to_address>
**Abandoned runs stay unreclaimed**
When a cached heap is built while a claim is live and that claim later expires or its owner dies, `TaskGenerator::heap()` reuses the heap while the checkpoint revision is unchanged, so it does not recheck `isAbandoned()`. The run is not handed out again by the long-lived scheduler until the heap is rebuilt.
Recheck claim abandonment as time passes or invalidate the cached heap when a claim can become reclaimable.
</issue_to_address>
### Comment 4
<location path="src/Generator/TaskGenerator.php" line_range="110" />
<code_context>
+ $this->claims[$task->getId()][] = $this->checkpoint->record($task->getId())
+ ?? [$this->resumeFrom[$task->getId()], TaskStatus::Done, null, null];
+
+ $this->checkpoint->write($task->getId(), $time, TaskStatus::Claimed, at: $now);
- // Reached whether the task succeeded or failed: the consumer keeps a failure to
</code_context>
<issue_to_address>
**Queued claims expire before start**
When the delay from draining a run to starting it exceeds the configured reclaim window, `TaskGenerator` timestamps every claim with the tick’s initial `$now`, before the concurrent scheduler waits for a free slot. If a queued run waits longer than `reclaimAfter()`, another tick reclaims it even though the original scheduler can later start it, so both schedulers run the task.
Start the reclaim window when a run actually begins, or ensure queued claims cannot be reclaimed while their scheduler still owns them.
</issue_to_address>
### Comment 5
<location path="src/TaskStatus.php" line_range="15-17" />
<code_context>
+ /**
+ * The run was handed out and what became of it is not known yet.
+ *
+ * A run still in flight reads as this, and so does one whose process died while it was
+ * going: the task is picked up after it either way, rather than risking a second run of
+ * something that may well have finished.
+ */
+ case Claimed = 'claimed';
</code_context>
<issue_to_address>
**Claim docs contradict replay behavior**
The `Claimed` PHPDoc says a run whose process died is picked up after its recorded time rather than risk a second run, but abandoned claims are rewound and handed out again. Consumers relying on the enum’s description are told the opposite of what the scheduler does.
Update the `Claimed` description to say that an abandoned claim is handed out again.
</issue_to_address>Sourcery assessment
Needs a human reviewer. 4 findings to address first, and if a claim outlives the reclaim window on another host, or where POSIX process checks are unavailable, the scheduler can hand the run out again even though the original task is still running; task side effects could therefore happen twice. Reverting stops future reclamation but cannot undo any duplicate external actions already taken.
Blocking findings: src/Generator/Checkpoint.php:165, src/Schedule.php:122, src/Generator/TaskGenerator.php:212, src/Generator/TaskGenerator.php:110
10e2adc to
8b8f799
Compare
|
Four of the five are in 8b8f799: completed runs becoming due again, live runs reclaimed immediately, abandoned runs staying unreclaimed, and the claim docs. Queued claims expiring before start is documented on |
|
Queued claims expire before start — 2d90e50. The claim is stamped as it is made rather than as the tick began, and |
2d90e50 to
7b4fbb7
Compare
samdark
left a comment
samdark
left a comment
There was a problem hiding this comment.
Two earlier concerns remain reproducible at 7b4fbb7, and returning work also breaks Scheduler reuse:
-
Late outcomes can still repeat completed work. Scheduler A starts 00:01, B completes 00:02, then A reaps its child.
resolve()checks A's stale in-memory record and writes 00:01 over 00:02, so the next scheduler repeats 00:02. This occurs with a shared cache and schedule mutex. Reload and validate the current shared record under the same exclusion protocol before outcome/start/release writes. -
Queued claims still expire before start. With a 10-second window, a 20-second delay before starting a queued task lets another scheduler reclaim and execute it; the original then executes it again. Resetting the window in
markStarted()happens after the vulnerable wait. Protect pending claims while waiting and verify current ownership before execution. -
Returning an unstarted run does not invalidate the cached heap (
TaskGenerator::release()). After graceful stop returns 00:01, the same Scheduler reports 00:02 as next and skips 00:01; a fresh Scheduler correctly runs 00:01. Advancing the reused Scheduler to 00:02 permanently replaces the returned occurrence with the new one. Rebuild pending work after release so reusing a stopped Scheduler preserves returned runs.
7b4fbb7 to
e7fad48
Compare
samdark
left a comment
samdark
left a comment
There was a problem hiding this comment.
The three previously reported defects still reproduce at e7fad48:
-
Late outcomes still overwrite newer shared state. After A claims 00:01 and B completes 00:02, A resolving its older run rewinds the checkpoint and another scheduler repeats 00:02. Reload and validate shared state under the same exclusion protocol before outcome/start/release writes.
-
Queued claims still expire before start. A 20-second slot wait with a 10-second reclaim window lets another scheduler execute the pending run before the original starts it too. Protect pending claims throughout the wait and verify current ownership before execution; renewing afterward is too late.
-
Returning an unstarted run in TaskGenerator::release() leaves the cached heap unchanged. After stopping, the reused Scheduler skips the returned 00:01 occurrence and reports 00:02, while a fresh Scheduler runs 00:01. Rebuild pending work after release so reuse preserves returned runs.
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.
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.
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.
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.
995d97f to
7628ee4
Compare
samdark
left a comment
samdark
left a comment
There was a problem hiding this comment.
Two issues remain at 7628ee4:
-
The new start guard drops missed runs. After a minute task misses 00:01, 00:02 and 00:03, concurrent generation leaves its single claim at 00:03.
markStarted()then rejects the legitimate 00:01 and 00:02 contexts. Withtick(true, 1), only 00:03 executes;e7fad48executes all three. Preserve earlier contexts from this scheduler's own catch-up while rejecting actual takeovers. -
Shared checkpoint updates still race. The reread fixes the ordering where another scheduler finished before the read, but validation and writing still lack the schedule mutex. If A reads its 00:01 claim, B completes 00:02, and A then writes its outcome, the checkpoint returns to 00:01 and a later scheduler repeats 00:02. Serialize the complete ownership-check and update with claiming, including the already-locked sequential path.
…le 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.
samdark
left a comment
samdark
left a comment
There was a problem hiding this comment.
At 27946aa, the README says the reclaim window applies only when the owner's process cannot be checked, but Checkpoint::isReclaimable() also expires claims whose same-host owner is positively alive. With reclaimAfter(10), I confirmed the owner and its running child were alive, advanced the injected clock by 20 seconds, and a second Scheduler executed the same 00:01 occurrence. The base executes it only once.
Please align the intended local-owner timeout contract, implementation, documentation and a public regression. If local liveness is meant to protect the claim, the unconditional timeout branch must respect that distinction. If all claims intentionally expire, document explicitly that a live local task can also be duplicated after the window. The general advice to choose a window longer than the task does not match the explicit local-owner exception currently documented. This requests a consistent contract rather than prescribing which policy to choose.
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.
|
Aligned the documentation to the window rather than the other way round — 767a136. A claim of a live local process expiring is deliberate: a pid is given to something else once its process is gone, so a claim waiting on one would stand forever, and asking about the process only ever brings the window forward. Regression in |
There was a problem hiding this comment.
One remaining documentation inconsistency
The multi-server section of README.md (lines 269-272) still states that "a task is still executed exactly once no matter how many servers the schedule runs on", without any condition. The "What became of a run" section (lines 225-234) changed by this commit says the opposite can happen: the reclaim window bounds every claim, including one whose process is still there, and taking a run back while its process still means to make it runs the task a second time. The code matches that text (Checkpoint::isReclaimable() expires a claim on elapsed time regardless of owner liveness).
Please qualify the multi-server exactly-once promise and disclose the duplicate-execution risk once a claim outlasts the reclaim window, instead of leaving it unconditional. (This asks for a consistent contract, implementation and documentation; it does not require the window to be the chosen mechanism.)
Nonblocking caveat: the new test covers only reissue of an unstarted claim held by a live process through the internal generator, not duplication of a running task through the public Scheduler; a public-path regression for the documented behavior is a reasonable follow-up.
This re-states no earlier finding (the ones from earlier heads remain addressed) and does not request code, test, or PR-description changes: a docs-wording fix only.
Stacked on #17, which is stacked on #16 — review those first. The base moves down as they land.
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. The concurrent tick drains the generator before starting anything, so a daemon told to stop left behind runs recorded as emitted that never ran. The code said so itself:
A run is now claimed as it is handed out and recorded as
done,failed,skippedorcancelledonce its outcome is known — the cases of the newTaskStatus. The runs a stopping daemon never gets to are given back and stay due.Claims that outlive their process
A claim carries the process that made it and the moment it was made.
posix_kill($pid, 0). The run it held is handed out again.ext-posix, cannot be asked about, so the claim is taken back only afterSchedule::reclaimAfter()— a day by default, and settable.Until then the run is left alone on purpose: handing out a run that is still going would run the task a second time, which is worse than picking one up late.
Two deliberate changes
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.
testTaskFailureIsNotMaskedByACheckpointPersistenceFailurecovered the old order and is replaced bytestRunIsNotHandedOutWhenItsClaimCannotBePersisted.In the sequential path the outcome is recorded after the run is over, so a checkpoint that cannot be written is still reported behind the failure of the task itself rather than instead of it.
What this sets up
onOneServerand per-task overlap prevention across hosts both need exactly this: a claim with an owner. So does a healthcheck, and aschedule:listthat can say when a task last succeeded.Summary by Sourcery
Track run ownership and outcomes so abandoned work can be safely reclaimed while in-flight and unstarted runs are not duplicated or lost.
New Features:
Bug Fixes:
Enhancements:
Documentation:
Tests: