From a21f8f3446ec0aadfe7f24ff4bcb6d4776dedf5e Mon Sep 17 00:00:00 2001 From: Dave Gaeddert Date: Fri, 10 Jul 2026 21:24:22 -0500 Subject: [PATCH] Replace pre-enqueued scheduled jobs with a ScheduleState ledger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The worker's scheduler used to enqueue each JOBS_SCHEDULE entry's next run as a future JobRequest, days ahead of its start time. Those materialized rows outlived the code that created them: removing or renaming a scheduled job class left a pending row that errored at pickup after the deploy, and changing an entry's timing let the old slot fire one last time. Scheduling is now evaluated at tick time against a ScheduleState ledger — one row per entry identity (job class + concurrency key + schedule), recording the last slot handled. Nothing is enqueued before its slot comes due, so removing an entry simply stops it from being evaluated. A slot is claimed with an optimistic UPDATE guarded by the previous ledger value, with the enqueue in the same transaction, which makes each slot fire exactly once across any number of workers with no leader election. Around the core: - A catch-up window (JOBS_SCHEDULE_CATCHUP_WINDOW, default 24h) replays the most recent missed slot after downtime while keeping a stale ledger row (an entry removed and re-added later) from firing a long-gone slot. The window also bounds the catch-up walk. - load_schedule validates entries fully (types, registration, cron ranges, duplicate identities) and a jobs.schedule preflight check reports every broken entry at deploy time. The admin gains a Schedule page that renders broken entries instead of failing. - JobClassNotRegistered replaces the bare KeyError when a stored job_class no longer resolves. - ScheduledCommand keys digest only when the raw command would overflow the 255-char concurrency_key with the slot stamp appended. - A PruneScheduleLedger chore deletes ledger rows for removed entries (queue-scoped workers can't tell those from another queue's). - Transition machinery for pre-ledger workers still running during the upgrade (a completed-run dedupe and a provenance-verified sweep of pre-enqueued future rows) is marked 'pre-ledger transition' and removable one release later. The ScheduleState unique constraint is created in the migration rather than convergence: workers get_or_create the same deterministic key at boot, and that race safety needs the constraint to exist before the first write. Claude-Session: https://claude.ai/code/session_01AEyVNMqpAhB1KVo6jrSPMk --- .claude/rules/plain-jobs.md | 1 + example/app/settings.py | 4 + plain-jobs/plain/jobs/README.md | 13 +- plain-jobs/plain/jobs/__init__.py | 4 +- plain-jobs/plain/jobs/admin.py | 56 ++ .../jobs/agents/.claude/rules/plain-jobs.md | 1 + plain-jobs/plain/jobs/chores.py | 23 +- plain-jobs/plain/jobs/default_settings.py | 9 +- plain-jobs/plain/jobs/exceptions.py | 19 + .../jobs/migrations/0002_schedulestate.py | 37 ++ plain-jobs/plain/jobs/models.py | 48 ++ plain-jobs/plain/jobs/preflight.py | 98 ++++ plain-jobs/plain/jobs/registry.py | 6 +- plain-jobs/plain/jobs/scheduling.py | 141 ++++- plain-jobs/plain/jobs/workers.py | 210 ++++++- plain-jobs/tests/internal/test_preflight.py | 114 ++++ .../tests/internal/test_scheduler_ledger.py | 531 ++++++++++++++++++ 17 files changed, 1265 insertions(+), 50 deletions(-) create mode 100644 plain-jobs/plain/jobs/migrations/0002_schedulestate.py create mode 100644 plain-jobs/plain/jobs/preflight.py create mode 100644 plain-jobs/tests/internal/test_preflight.py create mode 100644 plain-jobs/tests/internal/test_scheduler_ledger.py diff --git a/.claude/rules/plain-jobs.md b/.claude/rules/plain-jobs.md index 1f99e8fb03..d91ec48484 100644 --- a/.claude/rules/plain-jobs.md +++ b/.claude/rules/plain-jobs.md @@ -9,5 +9,6 @@ paths: - Offload slow work (email, API calls, file processing) to jobs — don't block HTTP responses - Use `concurrency_key` to prevent duplicate jobs for the same resource - Override `Job.on_aborted(result)` to clean up domain state when a job's process is terminated mid-run (status LOST or CANCELLED) — `run()`'s `try/finally` does not execute in that case +- Schedule recurring jobs with the `JOBS_SCHEDULE` setting — slots fire via a database ledger, so removing an entry just stops it (remove the entry when deleting a job class). `plain preflight` validates the schedule config Run `uv run plain docs jobs` for full documentation. diff --git a/example/app/settings.py b/example/app/settings.py index db7db6155b..446b16f6eb 100644 --- a/example/app/settings.py +++ b/example/app/settings.py @@ -32,6 +32,10 @@ "app.tasks", ] +JOBS_SCHEDULE = [ + ("app.jobs.ExampleJob", "*/5 * * * *"), +] + DEFAULT_RESPONSE_HEADERS = { # Strict CSP policy for testing CSP nonce support diff --git a/plain-jobs/plain/jobs/README.md b/plain-jobs/plain/jobs/README.md index cf8fbc5315..c9ae8480a4 100644 --- a/plain-jobs/plain/jobs/README.md +++ b/plain-jobs/plain/jobs/README.md @@ -153,10 +153,20 @@ Day-of-week follows standard cron numbering: **`0` (or `7`) is Sunday** through For custom schedules, see [`Schedule`](./scheduling.py#Schedule), which takes the same fields as keyword arguments. Its day-of-month and day-of-week combine with plain AND (both must match); pass `combine_days_with_or=True` for the cron OR behavior. +Workers evaluate the schedule a couple of times a minute and enqueue an entry when its slot comes due — nothing is enqueued ahead of time. Each entry has a [`ScheduleState`](./models.py#ScheduleState) ledger row recording the last slot it handled, which is what makes scheduling safe in the ways you'd hope: + +- Removing an entry just stops it from being evaluated. There is no pending future row left behind to error after a deploy. (When deleting a job class, remove its schedule entry too — an entry referencing a missing class fails at worker startup.) +- Changing an entry's timing takes effect immediately — the old timing can't fire one last time. +- A slot fires exactly once no matter how many workers are running, via an atomic ledger advance. +- If every worker is down when a slot passes, the most recent missed slot fires when one starts again — as long as the miss is within the catch-up window (`JOBS_SCHEDULE_CATCHUP_WINDOW`, default 24 hours). Older missed slots are skipped, so a stale ledger row (say, an entry removed and re-added weeks later) never fires a long-gone slot. The default favors completeness: a nightly rollup still runs after a deploy that spanned midnight, at the cost of a missed run firing late. If late runs are worse than skipped ones for your app (a morning digest arriving mid-afternoon), shrink the window. + +A new entry's ledger starts at deploy time, so nothing fires retroactively; the first run is the next slot after the entry is deployed. Note that `should_enqueue()` is evaluated when the slot comes due, so hooks that check current state (feature flags, tenant status) see the state at run time. + ## Admin interface The jobs package includes admin views for monitoring jobs under the "Jobs" section. The admin interface provides: +- **Schedule**: See `JOBS_SCHEDULE` entries with their next slot and the last one handled - **Requests**: View pending jobs in the queue - **Processes**: Monitor currently running jobs - **Results**: Review completed and failed job history @@ -243,7 +253,7 @@ Per-worker observable gauges (queryable per `messaging.destination.name` where a - `plain.jobs.worker.processes` — OS processes spawned by this worker - `plain.jobs.queue.depth` — pending `JobRequest`s ready to run -- `plain.jobs.queue.scheduled` — `JobRequest`s with `start_at` in the future +- `plain.jobs.queue.scheduled` — `JobRequest`s with `start_at` in the future (ad-hoc delayed enqueues and retry backoffs; cron-driven runs enqueue at their slot time, so they never appear here) - `plain.jobs.queue.oldest.age` — age in seconds of the oldest ready-to-run `JobRequest` - `plain.jobs.running` — `JobProcess` rows currently running @@ -266,6 +276,7 @@ Two contract details to be aware of: | `JOBS_HEARTBEAT_TIMEOUT` | `300` (5 minutes) | | `JOBS_MIDDLEWARE` | `[...]` | | `JOBS_SCHEDULE` | `[]` | +| `JOBS_SCHEDULE_CATCHUP_WINDOW` | `60 * 60 * 24` | | `JOBS_WORKER_MAX_PROCESSES` | `None` | | `JOBS_WORKER_MAX_JOBS_PER_PROCESS` | `None` | | `JOBS_WORKER_MAX_PENDING_PER_PROCESS` | `10` | diff --git a/plain-jobs/plain/jobs/__init__.py b/plain-jobs/plain/jobs/__init__.py index 6d0f4d41ec..8e3cc7e4c9 100644 --- a/plain-jobs/plain/jobs/__init__.py +++ b/plain-jobs/plain/jobs/__init__.py @@ -2,9 +2,9 @@ __version__ = version("plain.jobs") -from .exceptions import DeferJob +from .exceptions import DeferJob, JobClassNotRegistered from .jobs import Job from .middleware import JobMiddleware from .registry import register_job -__all__ = ["Job", "DeferJob", "JobMiddleware", "register_job"] +__all__ = ["Job", "DeferJob", "JobClassNotRegistered", "JobMiddleware", "register_job"] diff --git a/plain-jobs/plain/jobs/admin.py b/plain-jobs/plain/jobs/admin.py index 981289da8d..61b9cd2b81 100644 --- a/plain-jobs/plain/jobs/admin.py +++ b/plain-jobs/plain/jobs/admin.py @@ -1,13 +1,16 @@ from __future__ import annotations from datetime import timedelta +from typing import Any from plain import postgres from plain.admin.cards import Card, TrendCard from plain.admin.views import ( + AdminListView, AdminModelDetailView, AdminModelListView, AdminViewset, + register_view, register_viewset, ) from plain.http import RedirectResponse @@ -19,9 +22,15 @@ JobRequest, JobResult, JobResultQuerySet, + ScheduleState, WorkerHeartbeat, heartbeat_cutoff, ) +from .scheduling import ( + load_schedule_entry, + schedule_entry_display, + schedule_entry_key, +) def _td_format(td_object: timedelta) -> str: @@ -169,6 +178,53 @@ def get_link(self) -> str: return WorkerHeartbeatViewset.ListView.get_view_url() + "?display=Stale" +@register_view +class ScheduleView(AdminListView): + nav_section = "Jobs" + nav_icon = "calendar-week" + title = "Schedule" + description = "JOBS_SCHEDULE entries with their next slot and the last one handled." + path = "jobschedule/" + fields = ["job", "schedule", "queue", "next_slot", "last_enqueued_slot"] + + def get_initial_objects(self) -> list[dict[str, Any]]: + ledger = dict( + ScheduleState.query.values_list("schedule_key", "last_enqueued_slot") + ) + + rows = [] + for entry in settings.JOBS_SCHEDULE: + try: + job, schedule = load_schedule_entry(entry) + rows.append( + { + "job": schedule_entry_display(job), + "schedule": str(schedule), + "queue": job.default_queue(), + # next() raises for a schedule that can never match + "next_slot": schedule.next(), + "last_enqueued_slot": ledger.get( + schedule_entry_key(job, schedule) + ), + } + ) + except Exception as e: + # Render the broken entry instead of failing the whole page — + # this is exactly the state an operator is here to diagnose. + # repr(entry) is safe for any malformed shape. + rows.append( + { + "job": f"{entry!r} ({type(e).__name__})", + "schedule": "", + "queue": "", + "next_slot": None, + "last_enqueued_slot": None, + } + ) + + return rows + + @register_viewset class JobRequestViewset(AdminViewset): class ListView(AdminModelListView): diff --git a/plain-jobs/plain/jobs/agents/.claude/rules/plain-jobs.md b/plain-jobs/plain/jobs/agents/.claude/rules/plain-jobs.md index 1f99e8fb03..d91ec48484 100644 --- a/plain-jobs/plain/jobs/agents/.claude/rules/plain-jobs.md +++ b/plain-jobs/plain/jobs/agents/.claude/rules/plain-jobs.md @@ -9,5 +9,6 @@ paths: - Offload slow work (email, API calls, file processing) to jobs — don't block HTTP responses - Use `concurrency_key` to prevent duplicate jobs for the same resource - Override `Job.on_aborted(result)` to clean up domain state when a job's process is terminated mid-run (status LOST or CANCELLED) — `run()`'s `try/finally` does not execute in that case +- Schedule recurring jobs with the `JOBS_SCHEDULE` setting — slots fire via a database ledger, so removing an entry just stops it (remove the entry when deleting a job class). `plain preflight` validates the schedule config Run `uv run plain docs jobs` for full documentation. diff --git a/plain-jobs/plain/jobs/chores.py b/plain-jobs/plain/jobs/chores.py index d3df3d96e1..816d1e4ca4 100644 --- a/plain-jobs/plain/jobs/chores.py +++ b/plain-jobs/plain/jobs/chores.py @@ -4,7 +4,8 @@ from plain.runtime import settings from plain.utils import timezone -from .models import JobResult +from .models import JobResult, ScheduleState +from .scheduling import load_schedule, schedule_entry_key @register_chore @@ -17,3 +18,23 @@ def run(self) -> str: ) count = JobResult.query.filter(created_at__lt=cutoff).delete() return f"{count} jobs deleted" + + +@register_chore +class PruneScheduleLedger(Chore): + """Delete ScheduleState rows for entries no longer in JOBS_SCHEDULE. + + Queue-scoped workers only see their own entries, so they can't tell a + removed entry from another queue's — but chores run with the full + schedule config, making the deletion safe here. A row deleted out from + under a worker on a different config just re-initializes on its next + pass. + """ + + def run(self) -> str: + current_keys = [ + schedule_entry_key(job, schedule) + for job, schedule in load_schedule(settings.JOBS_SCHEDULE) + ] + count = ScheduleState.query.exclude(schedule_key__in=current_keys).delete() + return f"{count} stale schedule ledger rows deleted" diff --git a/plain-jobs/plain/jobs/default_settings.py b/plain-jobs/plain/jobs/default_settings.py index 332fc4a679..971eb4f1aa 100644 --- a/plain-jobs/plain/jobs/default_settings.py +++ b/plain-jobs/plain/jobs/default_settings.py @@ -6,7 +6,14 @@ JOBS_MIDDLEWARE: list[str] = [ "plain.jobs.middleware.AppLoggerMiddleware", ] -JOBS_SCHEDULE: list[tuple[str, str]] = [] +# Entries are (job, schedule) tuples: a dotted job-class path or "cmd:" string +# (or a Job instance), and a cron string (or a Schedule instance). +JOBS_SCHEDULE: list[tuple] = [] +# How far back (in seconds) a missed slot may be and still fire when a +# worker returns — catch-up after ordinary downtime without a stale ledger +# row firing a long-gone slot. Floored at two minutes: anything smaller +# would skip slots that come due between ordinary scheduler passes. +JOBS_SCHEDULE_CATCHUP_WINDOW: int = 60 * 60 * 24 JOBS_WORKER_MAX_PROCESSES: int | None = None JOBS_WORKER_MAX_JOBS_PER_PROCESS: int | None = None JOBS_WORKER_MAX_PENDING_PER_PROCESS: int = 10 diff --git a/plain-jobs/plain/jobs/exceptions.py b/plain-jobs/plain/jobs/exceptions.py index a9aab134f2..04e3be0c78 100644 --- a/plain-jobs/plain/jobs/exceptions.py +++ b/plain-jobs/plain/jobs/exceptions.py @@ -1,3 +1,22 @@ +from __future__ import annotations + + +class JobClassNotRegistered(Exception): + """A job_class name has no registered Job class. + + Usually means the class was renamed or removed in a deploy while + something still references it — pending database rows (requests, + retries, results) or a JOBS_SCHEDULE entry. + """ + + def __init__(self, job_class_name: str): + self.job_class_name = job_class_name + super().__init__( + f"Job class '{job_class_name}' is not registered — it may have " + "been renamed or removed in a deploy." + ) + + class DeferJob(Exception): """Signal that a job should be deferred and re-tried later. diff --git a/plain-jobs/plain/jobs/migrations/0002_schedulestate.py b/plain-jobs/plain/jobs/migrations/0002_schedulestate.py new file mode 100644 index 0000000000..460c383c46 --- /dev/null +++ b/plain-jobs/plain/jobs/migrations/0002_schedulestate.py @@ -0,0 +1,37 @@ +# Generated by Plain 0.151.1 on 2026-07-10 21:29 + +from plain import postgres +from plain.postgres import migrations + + +class Migration(migrations.Migration): + dependencies = [ + ("plainjobs", "0001_initial"), + ] + + operations = [ + migrations.CreateModel( + name="ScheduleState", + fields=[ + ("id", postgres.PrimaryKeyField()), + ("created_at", postgres.DateTimeField(create_now=True)), + ("last_enqueued_slot", postgres.DateTimeField()), + ("schedule_key", postgres.TextField()), + ], + options={ + "ordering": ["schedule_key"], + }, + ), + # Constraints normally arrive via schema convergence, but workers + # get_or_create ledger rows as soon as they boot — before a converge + # step may have run — and get_or_create's race safety depends on this + # unique constraint existing. Create it here so there is no window + # where duplicate schedule_key rows can be inserted (which would then + # block convergence from ever adding the constraint). Convergence + # sees the matching name/definition and leaves it alone. + migrations.RunSQL( + 'ALTER TABLE "plainjobs_schedulestate"' + ' ADD CONSTRAINT "plainjobs_schedulestate_unique_schedule_key"' + ' UNIQUE ("schedule_key")' + ), + ] diff --git a/plain-jobs/plain/jobs/models.py b/plain-jobs/plain/jobs/models.py index 69e3dbe4fe..e837db1f6d 100644 --- a/plain-jobs/plain/jobs/models.py +++ b/plain-jobs/plain/jobs/models.py @@ -41,6 +41,7 @@ "JobResult", "JobResultStatuses", "WorkerHeartbeat", + "ScheduleState", ] logger = get_framework_logger() @@ -621,6 +622,12 @@ class JobResult(postgres.Model): name="plainjobs_jobresult_created_at_idx", fields=["created_at"] ), postgres.Index(name="plainjobs_jobresult_status_idx", fields=["status"]), + # Used by the scheduler's completed-slot dedupe, and matches the + # equivalent index on JobRequest/JobProcess. + postgres.Index( + name="job_result_concurrency_key", + fields=["job_class", "concurrency_key"], + ), ], constraints=[ postgres.UniqueConstraint( @@ -726,6 +733,47 @@ def __str__(self) -> str: return f"WorkerHeartbeat({self.worker_id} on {self.hostname}:{self.pid})" +@postgres.register_model +class ScheduleState(postgres.Model): + """ + The scheduler's ledger: one row per JOBS_SCHEDULE entry, recording the + last slot handled for it. A slot is handled when its job is enqueued — + or when it's settled without firing: seen for the first time at deploy, + skipped by should_enqueue() or the completed-run dedupe, or older than + the catch-up window. + + Workers evaluate schedules at tick time — nothing is enqueued ahead of + its start time — so removing an entry (or its job class) from + JOBS_SCHEDULE just stops it from being evaluated. A row whose entry no + longer exists is inert bookkeeping, not pending work. + + Advancing `last_enqueued_slot` uses an optimistic + `UPDATE ... WHERE last_enqueued_slot = ` so an entry fires at most + once per slot across any number of workers. + """ + + # No max_length: the key embeds the schedule and, for ScheduledCommand, + # the full shell command — a length limit here would make one long entry + # fail validation on every scheduling pass. + schedule_key = types.TextField() + last_enqueued_slot = types.DateTimeField() + created_at = types.DateTimeField(create_now=True) + + model_options = postgres.Options( + ordering=["schedule_key"], + constraints=[ + # The unique constraint provides the schedule_key lookup index. + postgres.UniqueConstraint( + fields=["schedule_key"], + name="plainjobs_schedulestate_unique_schedule_key", + ), + ], + ) + + def __str__(self) -> str: + return f"ScheduleState({self.schedule_key} @ {self.last_enqueued_slot})" + + def heartbeat_cutoff() -> datetime.datetime: """The timestamp before which a WorkerHeartbeat is considered stale. diff --git a/plain-jobs/plain/jobs/preflight.py b/plain-jobs/plain/jobs/preflight.py new file mode 100644 index 0000000000..29627491e1 --- /dev/null +++ b/plain-jobs/plain/jobs/preflight.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from plain.preflight import PreflightCheck, PreflightResult, register_check +from plain.runtime import settings +from plain.utils import timezone + +from .exceptions import JobClassNotRegistered +from .scheduling import ( + load_schedule_entry, + schedule_entry_display, + schedule_entry_key, + scheduled_concurrency_key, +) + + +@register_check(name="jobs.schedule") +class CheckJobsSchedule(PreflightCheck): + """ + JOBS_SCHEDULE must load and every entry must be runnable: registered job + classes, valid schedules that can actually match, distinct identities, + and concurrency keys short enough to enqueue. The worker refuses to boot + (or an entry silently never fires) otherwise, so catching it here stops + the bad config at deploy time. Entries are checked individually so one + broken entry doesn't hide the others. + """ + + def run(self) -> list[PreflightResult]: + results = [] + loaded = [] + + for entry in settings.JOBS_SCHEDULE: + try: + loaded.append(load_schedule_entry(entry)) + except JobClassNotRegistered as e: + results.append( + PreflightResult( + fix=f"{e} Remove the JOBS_SCHEDULE entry or restore the class.", + id="jobs.schedule.unregistered_class", + ) + ) + except Exception as e: + # Wrong entry shape, malformed cron string, out-of-range + # field — the message says which. + results.append( + PreflightResult( + fix=f"JOBS_SCHEDULE entry {entry!r} is invalid: {e}", + id="jobs.schedule.invalid", + ) + ) + + seen_keys: set[str] = set() + + for job, schedule in loaded: + key = schedule_entry_key(job, schedule) + if key in seen_keys: + results.append( + PreflightResult( + fix=( + f"Duplicate JOBS_SCHEDULE entry: {key!r}. Entries " + "with the same job class and schedule need distinct " + "default_concurrency_key() values." + ), + id="jobs.schedule.duplicate_entry", + ) + ) + seen_keys.add(key) + + try: + schedule.next() + except ValueError: + # A warning, not an error: next() scans 500 days, so a valid + # sparse schedule (e.g. Feb 29) can land here depending on + # the date preflight runs — that must not block a deploy. + results.append( + PreflightResult( + fix=( + f"JOBS_SCHEDULE entry {schedule_entry_display(job)!r} " + "does not match any time in the next 500 days — " + "check the schedule." + ), + id="jobs.schedule.never_matches", + warning=True, + ) + ) + + if len(scheduled_concurrency_key(job, timezone.now())) > 255: + results.append( + PreflightResult( + fix=( + f"JOBS_SCHEDULE entry {schedule_entry_display(job)!r} " + "has a default_concurrency_key() too long to enqueue " + "(concurrency_key is limited to 255 characters)." + ), + id="jobs.schedule.concurrency_key_too_long", + ) + ) + + return results diff --git a/plain-jobs/plain/jobs/registry.py b/plain-jobs/plain/jobs/registry.py index d0493e0cca..558850d9d0 100644 --- a/plain-jobs/plain/jobs/registry.py +++ b/plain-jobs/plain/jobs/registry.py @@ -3,6 +3,7 @@ from collections.abc import Callable from typing import TYPE_CHECKING, Any, overload +from .exceptions import JobClassNotRegistered from .parameters import JobParameters if TYPE_CHECKING: @@ -25,7 +26,10 @@ def get_job_class_name(self, job_class: type[Job]) -> str: return f"{job_class.__module__}.{job_class.__qualname__}" def get_job_class(self, name: str) -> type[Job]: - return self.jobs[name] + try: + return self.jobs[name] + except KeyError: + raise JobClassNotRegistered(name) from None def load_job( self, job_class_name: str, parameters: dict[str, Any] | None = None diff --git a/plain-jobs/plain/jobs/scheduling.py b/plain-jobs/plain/jobs/scheduling.py index 92401c14a8..b5c600d5b1 100644 --- a/plain-jobs/plain/jobs/scheduling.py +++ b/plain-jobs/plain/jobs/scheduling.py @@ -1,11 +1,13 @@ from __future__ import annotations import datetime +import hashlib import subprocess from typing import Any from plain.utils import timezone +from .exceptions import JobClassNotRegistered from .jobs import Job from .registry import jobs_registry, register_job @@ -102,6 +104,17 @@ def _convert(value: str) -> int: result = str_conversions.get(value.upper(), value) return int(result) + def _validated(values: list[int]) -> list[int]: + # String forms ("25", "20-30") need the same bounds check as + # plain ints — an out-of-range value would otherwise only blow + # up later, inside Schedule.next()'s datetime arithmetic. + for v in values: + if v < min_allowed or v > max_allowed: + raise ValueError( + f"Schedule component should be between {min_allowed} and {max_allowed}" + ) + return values + if "/" in value: values, step = value.split("/") values = cls.parse(values, min_allowed, max_allowed, str_conversions) @@ -109,9 +122,11 @@ def _convert(value: str) -> int: if "-" in value: start, end = value.split("-") - return cls(list(range(_convert(start), _convert(end) + 1)), raw=value) + return cls( + _validated(list(range(_convert(start), _convert(end) + 1))), raw=value + ) - return cls([_convert(value)], raw=value) + return cls(_validated([_convert(value)]), raw=value) class Schedule: @@ -162,7 +177,16 @@ def __init__( def __str__(self) -> str: if self._raw: return self._raw - return f"{self.minute} {self.hour} {self.day_of_month} {self.month} {self.day_of_week}" + fields = f"{self.minute} {self.hour} {self.day_of_month} {self.month} {self.day_of_week}" + if ( + self.combine_days_with_or + and not self.day_of_month.is_wildcard + and not self.day_of_week.is_wildcard + ): + # The OR day-combination matches a different set of slots, so + # it's part of the schedule's identity (and ledger key). + fields += " (days OR)" + return fields def __repr__(self) -> str: return f"" @@ -272,24 +296,111 @@ def run(self) -> None: def default_concurrency_key(self) -> str: # The ScheduledCommand can be used for different commands, - # so we need the concurrency_key to separate them for uniqueness - return self.command + # so we need the concurrency_key to separate them for uniqueness. + # Digest only when the raw command wouldn't fit + # JobRequest.concurrency_key's 255-char bound once the + # ":scheduled:" slot stamp (21 chars) is appended — a command + # that fits keeps its raw key, which also matches the keys + # pre-ledger workers produced so slot dedupe holds across the + # upgrade. + if len(self.command) <= 234: + return self.command + digest = hashlib.sha256(self.command.encode()).hexdigest()[:16] + return f"{self.command[:160]}...{digest}" + + +def schedule_entry_key(job: Job, schedule: Schedule) -> str: + """ + Stable identity of a JOBS_SCHEDULE entry, used as ScheduleState.schedule_key. + + Includes the schedule itself, so changing an entry's timing starts a + fresh ledger row (the old one becomes inert) and two entries for the + same job with different schedules track independently. + """ + job_class_name = jobs_registry.get_job_class_name(job.__class__) + return f"{job_class_name}:{job.default_concurrency_key()}:{schedule}" + + +def schedule_entry_display(job: Job) -> str: + """How a JOBS_SCHEDULE entry names itself — the inverse of load_schedule's + parsing, so the `cmd:` convention lives in one module.""" + if isinstance(job, ScheduledCommand): + return f"cmd:{job.command}" + return jobs_registry.get_job_class_name(job.__class__) + + +def scheduled_concurrency_key(job: Job, slot: datetime.datetime) -> str: + """The concurrency_key stamped on a scheduled run's rows. + + Groups a slot's request/process/result rows for queries, and its + uniqueness under should_enqueue() dedupes against a pending row another + process already created for the same slot (e.g. one pre-enqueued before + the upgrade to ledger-based scheduling — the sweep migration selects + legacy rows by this format's `:scheduled:` marker). + """ + return f"{job.default_concurrency_key()}:scheduled:{int(slot.timestamp())}" + + +def load_schedule_entry( + entry: tuple[str | Job, str | Schedule], +) -> tuple[Job, Schedule]: + """Parse a single JOBS_SCHEDULE entry — raises if anything about it is + wrong (shape, unregistered class, malformed schedule).""" + job, schedule = entry + + if isinstance(job, str): + if job.startswith("cmd:"): + job = ScheduledCommand(job[4:]) + else: + job = jobs_registry.load_job(job, {"args": [], "kwargs": {}}) + + if isinstance(schedule, str): + schedule = Schedule.from_cron(schedule) + + # The settings type only guarantees tuples, so validate what came out. + if not isinstance(job, Job): + raise ValueError( + f"JOBS_SCHEDULE job must be a Job class path, cmd: string, or " + f"Job instance — got {job!r}" + ) + if not isinstance(schedule, Schedule): + raise ValueError( + f"JOBS_SCHEDULE schedule must be a cron string or Schedule — " + f"got {schedule!r}" + ) + + # A Job instance whose class was never registered would enqueue rows + # that can't be loaded at pickup. + job_class_name = jobs_registry.get_job_class_name(job.__class__) + if job_class_name not in jobs_registry.jobs: + raise JobClassNotRegistered(job_class_name) + + return job, schedule def load_schedule( schedules: list[tuple[str | Job, str | Schedule]], ) -> list[tuple[Job, Schedule]]: + """Parse JOBS_SCHEDULE, failing on the first problem — the worker + refuses to boot on broken schedule config. Consumers that want to report + every problem (preflight) or render broken entries (admin) iterate with + load_schedule_entry instead.""" jobs_schedule: list[tuple[Job, Schedule]] = [] - - for job, schedule in schedules: - if isinstance(job, str): - if job.startswith("cmd:"): - job = ScheduledCommand(job[4:]) - else: - job = jobs_registry.load_job(job, {"args": [], "kwargs": {}}) - - if isinstance(schedule, str): - schedule = Schedule.from_cron(schedule) + seen_keys: set[str] = set() + + for entry in schedules: + job, schedule = load_schedule_entry(entry) + + # Entries with the same key would share one ledger row and only one + # of them would ever fire — refuse loudly instead. + key = schedule_entry_key(job, schedule) + if key in seen_keys: + raise ValueError( + f"Duplicate JOBS_SCHEDULE entry: {key!r}. Entries with the " + "same job class and schedule need distinct " + "default_concurrency_key() values to be scheduled separately." + ) + seen_keys.add(key) jobs_schedule.append((job, schedule)) diff --git a/plain-jobs/plain/jobs/workers.py b/plain-jobs/plain/jobs/workers.py index 1c8896cfd5..dd6bcb9879 100644 --- a/plain-jobs/plain/jobs/workers.py +++ b/plain-jobs/plain/jobs/workers.py @@ -27,8 +27,10 @@ from .otel import WorkerMetrics, emit_error_consumer_span, record_span_error, tracer from .registry import jobs_registry +from .scheduling import Schedule, schedule_entry_key, scheduled_concurrency_key if TYPE_CHECKING: + from .jobs import Job from .models import JobProcess, JobResult # Models are NOT imported at the top of this file! @@ -108,7 +110,10 @@ def __init__( now = time.time() self._stats_logged_at = now self._job_results_checked_at = now - self._jobs_schedule_checked_at = now + # Except the schedule: evaluate it on the first tick so a slot that + # came due while no worker was running fires at startup, not a + # minute in. + self._jobs_schedule_checked_at = 0.0 self.worker_id = uuid.uuid4() self._hostname = socket.gethostname() @@ -495,45 +500,192 @@ def deregister_heartbeat(self) -> None: logger.exception(e) def _schedule_due(self, now: float) -> bool: - if not self.jobs_schedule: - return False - # Only need to check once every 60 seconds - return now - self._jobs_schedule_checked_at > 60 + # Pre-ledger transition: the pass must run even when this worker has + # no schedule entries, so the legacy sweep still cleans rows an old + # worker re-created after the last entry was removed. Restore + # `if not self.jobs_schedule: return False` here when the transition + # machinery is removed. + # + # Passes must run more often than the smallest schedule period (one + # minute), or two slots could come due within a single gap and the + # latest-only catch-up below would silently drop the earlier one. + return now - self._jobs_schedule_checked_at > 30 def maybe_schedule_jobs(self) -> None: + """Enqueue any schedule entry whose slot has come due. + + Nothing is enqueued ahead of time — see ScheduleState for how the + ledger makes each slot fire exactly once and keeps removed entries + from leaving stale work behind. + """ if not self._schedule_due(time.time()): return + # Stamp at the start: a slot can't be lost to a delayed stamp (the + # ledger owns slots), and a failing entry is contained per-entry + # below rather than re-running the whole pass every loop tick. + self._jobs_schedule_checked_at = time.time() + + # Lazy import - see _worker_process_initializer() comment for why + from .models import ScheduleState + + now = timezone.localtime() + + self._sweep_legacy_scheduled_requests(now) + + if not self.jobs_schedule: + return - for job, schedule in self.jobs_schedule: - next_start_at = schedule.next() + entries = [ + (job, schedule, schedule_entry_key(job, schedule)) + for job, schedule in self.jobs_schedule + ] + + ledger = dict( + ScheduleState.query.filter( + schedule_key__in=[key for _, _, key in entries] + ).values_list("schedule_key", "last_enqueued_slot") + ) + + for job, schedule, schedule_key in entries: + try: + self._schedule_entry( + job, schedule, schedule_key, ledger.get(schedule_key), now + ) + except Exception as e: + # One broken entry must not starve the entries after it — but + # the catch also stops the SDK auto-record, so stamp the + # canonical failure signal on the active worker-loop span or + # APM would show a healthy worker while an entry never runs. + record_span_error(trace.get_current_span(), e) + logger.exception(e) + + def _sweep_legacy_scheduled_requests(self, now: datetime.datetime) -> None: + """Pre-ledger transition: delete rows pre-enqueued by a pre-ledger + worker (before the upgrade, or re-created by one still running + during a rolling deploy). + + Only future rows are swept — one whose slot already passed is a run + the old scheduler owed and still executes at pickup, preserving its + catch-up behavior through the upgrade. This scheduler never creates + future-dated slot rows (it enqueues only due slots), so any future + row whose key's trailing digits equal its own start time's epoch — + the old format's provenance — is legacy by definition. Removable + once pre-ledger workers can no longer be running. + """ + # Lazy import - see _worker_process_initializer() comment for why + from .models import JobRequest - # Leverage the concurrency_key to group scheduled jobs - # with the same start time - schedule_concurrency_key = f"{job.default_concurrency_key()}:scheduled:{int(next_start_at.timestamp())}" + candidates = JobRequest.query.filter( + concurrency_key__regex=r":scheduled:\d+$", + retry_attempt=0, + start_at__gt=now, + ).values_list("id", "concurrency_key", "start_at") + + ids = [ + id + for id, concurrency_key, start_at in candidates + if int(concurrency_key.rsplit(":", 1)[1]) == int(start_at.timestamp()) + ] + + if ids: + deleted = JobRequest.query.filter(id__in=ids).delete() + logger.info( + "Swept legacy pre-enqueued scheduled requests", + extra={"deleted": deleted}, + ) + + def _schedule_entry( + self, + job: Job, + schedule: Schedule, + schedule_key: str, + last_enqueued_slot: datetime.datetime | None, + now: datetime.datetime, + ) -> None: + # Lazy import - see _worker_process_initializer() comment for why + from .models import JobResult, ScheduleState + + if last_enqueued_slot is None: + # First time this entry is seen: the ledger starts at now, so + # nothing fires retroactively — the first slot to fire is the + # next one after deployment. + ScheduleState.query.get_or_create( + schedule_key=schedule_key, + defaults={"last_enqueued_slot": now}, + ) + return + + last_slot = timezone.localtime(last_enqueued_slot) + if schedule.next(now=last_slot) > now: + return + + # A slot is due. Fire the latest one, but only if it falls inside + # the catch-up window — a worker returning from ordinary downtime + # replays the most recent miss, while a stale ledger row doesn't + # fire a long-gone slot. Walking from the window edge also bounds + # the loop over large gaps. + # Floored at two minutes: a window smaller than the gap between + # passes would classify slots that came due between two ordinary + # passes as stale and skip them — i.e. disable scheduling. + window = datetime.timedelta( + seconds=max(settings.JOBS_SCHEDULE_CATCHUP_WINDOW, 120) + ) + due_slot = schedule.next(now=max(last_slot, now - window)) + if due_slot > now: + # The only missed slots are older than the window. Advance the + # ledger without firing so they aren't re-evaluated every pass. + ScheduleState.query.filter( + schedule_key=schedule_key, + last_enqueued_slot=last_enqueued_slot, + ).update(last_enqueued_slot=now) + return + while (following := schedule.next(now=due_slot)) <= now: + due_slot = following + + # Optimistically claim the slot: the ledger only advances if no + # other worker got there first, and the enqueue shares the + # transaction so a failed enqueue releases the slot to retry next + # pass. + with transaction.atomic(): + claimed = ScheduleState.query.filter( + schedule_key=schedule_key, + last_enqueued_slot=last_enqueued_slot, + ).update(last_enqueued_slot=due_slot) + if not claimed: + return + + concurrency_key = scheduled_concurrency_key(job, due_slot) + + # Pre-ledger transition: a run for this slot that already + # completed leaves only a JobResult, which should_enqueue's + # pending/processing dedupe can't see — e.g. a row pre-enqueued + # by a pre-ledger worker during a rolling upgrade that ran + # before this pass. Keep the ledger advance, skip the enqueue. + # Removable once pre-ledger workers can no longer be running. + if JobResult.query.filter( + job_class=jobs_registry.get_job_class_name(job.__class__), + concurrency_key=concurrency_key, + ).exists(): + return # Job's should_enqueue hook can control scheduling behavior result = job.run_in_worker( - delay=next_start_at, - concurrency_key=schedule_concurrency_key, + delay=due_slot, + concurrency_key=concurrency_key, ) - # Result is None if should_enqueue returned False - if result: - logger.info( - "Scheduling job", - extra={ - "job_class": result.job_class, - "job_queue": result.queue, - "job_start_at": result.start_at, - "job_schedule": schedule, - "concurrency_key": result.concurrency_key, - }, - ) - # Stamp only after the whole pass succeeds — a mid-pass failure - # (caught by the worker-loop catch) then retries next tick instead - # of waiting out the window and skipping the missed occurrence. - # Re-runs are deduped by the scheduled concurrency_key. - self._jobs_schedule_checked_at = time.time() + # Result is None if should_enqueue returned False + if result: + logger.info( + "Scheduling job", + extra={ + "job_class": result.job_class, + "job_queue": result.queue, + "job_start_at": result.start_at, + "job_schedule": schedule, + "concurrency_key": result.concurrency_key, + }, + ) def log_stats(self) -> None: # Lazy import - see _worker_process_initializer() comment for why diff --git a/plain-jobs/tests/internal/test_preflight.py b/plain-jobs/tests/internal/test_preflight.py new file mode 100644 index 0000000000..bfc131c4b1 --- /dev/null +++ b/plain-jobs/tests/internal/test_preflight.py @@ -0,0 +1,114 @@ +"""The jobs.schedule preflight check: every way JOBS_SCHEDULE can be broken +is reported at deploy time, per entry, without hiding the others.""" + +from __future__ import annotations + +from plain.jobs import Job +from plain.jobs.preflight import CheckJobsSchedule +from plain.jobs.registry import jobs_registry, register_job + + +@register_job +class _PreflightJob(Job): + def run(self) -> None: + pass + + +@register_job +class _LongKeyJob(Job): + def run(self) -> None: + pass + + def default_concurrency_key(self) -> str: + return "x" * 300 + + +class _UnregisteredJob(Job): + """Deliberately not @register_job'd.""" + + def run(self) -> None: + pass + + +JOB = jobs_registry.get_job_class_name(_PreflightJob) + + +def _check(settings, schedule) -> list[str]: + settings.JOBS_SCHEDULE = schedule + return [result.id for result in CheckJobsSchedule().run()] + + +def test_valid_schedule_passes(settings): + assert _check(settings, [(JOB, "@daily"), ("cmd:echo ok", "@hourly")]) == [] + + +def test_unregistered_class(settings): + assert _check(settings, [("app.gone.RemovedJob", "@daily")]) == [ + "jobs.schedule.unregistered_class" + ] + + +def test_malformed_cron_string(settings): + assert _check(settings, [(JOB, "* * * *")]) == ["jobs.schedule.invalid"] + + +def test_out_of_range_cron_field_is_blocking(settings): + # Range validation happens at parse time for string fields too — this + # must be a blocking `invalid`, not the nonblocking never-matches + # warning (or a worker that error-loops on every pass). + assert _check(settings, [(JOB, "0 25 * * *")]) == ["jobs.schedule.invalid"] + assert _check(settings, [(JOB, "0 20-30 * * *")]) == ["jobs.schedule.invalid"] + + +def test_malformed_entry_shape(settings): + # A non-tuple entry is rejected by settings typing before preflight ever + # sees it; a wrong-arity tuple gets through and is reported here. + assert _check(settings, [(JOB,)]) == ["jobs.schedule.invalid"] + + +def test_wrong_member_types(settings): + # Members of the tuple aren't settings-validated — a None schedule or a + # non-Job object must be reported, not crash the check. + assert _check(settings, [(_PreflightJob(), None)]) == ["jobs.schedule.invalid"] + assert _check(settings, [(object(), "@daily")]) == ["jobs.schedule.invalid"] + + +def test_unregistered_job_instance(settings): + # An instance loads fine but its rows could never be picked up. + assert _check(settings, [(_UnregisteredJob(), "@daily")]) == [ + "jobs.schedule.unregistered_class" + ] + + +def test_duplicate_entries(settings): + assert _check(settings, [(JOB, "@daily"), (JOB, "@daily")]) == [ + "jobs.schedule.duplicate_entry" + ] + + +def test_schedule_that_never_matches_warns(settings): + # A warning, not an error — a valid sparse schedule (e.g. Feb 29) can + # exceed next()'s 500-day scan depending on the deploy date, and that + # must not block a deploy. + settings.JOBS_SCHEDULE = [(JOB, "0 0 30 2 *")] + results = CheckJobsSchedule().run() + assert [result.id for result in results] == ["jobs.schedule.never_matches"] + assert results[0].warning + + +def test_concurrency_key_too_long(settings): + assert _check(settings, [(_LongKeyJob(), "@daily")]) == [ + "jobs.schedule.concurrency_key_too_long" + ] + + +def test_broken_entries_are_all_reported(settings): + ids = _check( + settings, + [ + ("app.gone.RemovedJob", "@daily"), + (JOB, "* * * *"), + (JOB, "@daily"), # valid — doesn't mask or get masked + ], + ) + assert ids == ["jobs.schedule.unregistered_class", "jobs.schedule.invalid"] diff --git a/plain-jobs/tests/internal/test_scheduler_ledger.py b/plain-jobs/tests/internal/test_scheduler_ledger.py new file mode 100644 index 0000000000..99852fe2f8 --- /dev/null +++ b/plain-jobs/tests/internal/test_scheduler_ledger.py @@ -0,0 +1,531 @@ +"""Ledger-based schedule evaluation: entries fire once per due slot, never +early, never retroactively, and stop firing the moment they leave +JOBS_SCHEDULE.""" + +from __future__ import annotations + +import datetime +import uuid + +import pytest + +from plain.jobs import Job +from plain.jobs.models import ( + JobProcess, + JobRequest, + JobResult, + JobResultStatuses, + ScheduleState, +) +from plain.jobs.registry import jobs_registry, register_job +from plain.jobs.scheduling import ( + Schedule, + ScheduledCommand, + load_schedule, + schedule_entry_key, + scheduled_concurrency_key, +) +from plain.jobs.workers import Worker +from plain.utils import timezone + +EVERY_MINUTE = Schedule.from_cron("* * * * *") + + +@register_job +class _TickJob(Job): + def run(self) -> None: + pass + + +@register_job +class _NeverEnqueueJob(Job): + def run(self) -> None: + pass + + def should_enqueue(self, concurrency_key: str) -> bool: + return False + + +@register_job +class _RaisingEnqueueJob(Job): + def run(self) -> None: + pass + + def should_enqueue(self, concurrency_key: str) -> bool: + raise RuntimeError("boom from should_enqueue") + + +@pytest.fixture +def make_worker(): + """Build Workers whose schedule pass runs immediately when called.""" + workers = [] + + def _make(jobs_schedule=None) -> Worker: + worker = Worker( + queues=["default"], + jobs_schedule=jobs_schedule or [], + max_processes=1, + ) + workers.append(worker) + return worker + + yield _make + + for worker in workers: + worker.executor.shutdown(wait=False, cancel_futures=True) + + +def _backdate_ledger(job: Job, schedule: Schedule, *, minutes: int) -> None: + """Move an entry's ledger into the past so a slot is due right now.""" + ScheduleState.query.filter(schedule_key=schedule_entry_key(job, schedule)).update( + last_enqueued_slot=timezone.now() - datetime.timedelta(minutes=minutes) + ) + + +def _run_schedule_pass(worker: Worker) -> None: + """Run another schedule pass now, bypassing the schedule gate (the boot + pass already consumed the immediate-evaluation baseline).""" + worker._jobs_schedule_checked_at = 0.0 + worker.maybe_schedule_jobs() + + +def test_first_pass_initializes_ledger_without_firing(db, make_worker): + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + + worker.maybe_schedule_jobs() + + state = ScheduleState.query.get(schedule_key=schedule_entry_key(job, EVERY_MINUTE)) + assert state.last_enqueued_slot <= timezone.now() + assert not JobRequest.query.exists() + + +def test_due_slot_enqueues_ready_to_run_job(db, make_worker): + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + _run_schedule_pass(worker) + + job_request = JobRequest.query.get() + assert job_request.job_class == jobs_registry.get_job_class_name(_TickJob) + start_at = job_request.start_at + assert start_at is not None + assert start_at <= timezone.now() + # start_at is the slot time itself (a minute boundary), not "now" + assert start_at.second == 0 + assert start_at.microsecond == 0 + assert JobRequest.query.ready_to_run().exists() + assert ":scheduled:" in job_request.concurrency_key + + +def test_slot_fires_once_across_worker_restarts(db, make_worker): + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + _run_schedule_pass(worker) + + # A restarted worker (fresh instance, same schedule) sees the advanced + # ledger and enqueues nothing new. + restarted = make_worker([(_TickJob(), EVERY_MINUTE)]) + restarted.maybe_schedule_jobs() + + assert JobRequest.query.count() == 1 + + +def test_two_workers_fire_a_slot_once(db, make_worker): + job_a, job_b = _TickJob(), _TickJob() + worker_a = make_worker([(job_a, EVERY_MINUTE)]) + worker_b = make_worker([(job_b, EVERY_MINUTE)]) + worker_a.maybe_schedule_jobs() + _backdate_ledger(job_a, EVERY_MINUTE, minutes=2) + + _run_schedule_pass(worker_a) + worker_b.maybe_schedule_jobs() + + assert JobRequest.query.count() == 1 + + +def test_missed_slots_fire_latest_only(db, make_worker): + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=30) + + _run_schedule_pass(worker) + + job_request = JobRequest.query.get() + # Only the most recent due slot fired, not the ~30 missed ones. + start_at = job_request.start_at + assert start_at is not None + assert timezone.now() - start_at < datetime.timedelta(seconds=60) + + +def test_removed_entry_stops_firing_and_leaves_nothing_pending(db, make_worker): + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + # The entry was removed from JOBS_SCHEDULE (worker restarts without it) + # before its slot came due. Nothing fires and nothing is pending — the + # ledger row is inert bookkeeping. + without_entry = make_worker([]) + without_entry.maybe_schedule_jobs() + + assert not JobRequest.query.exists() + assert ScheduleState.query.count() == 1 + + +def test_changed_schedule_starts_a_fresh_ledger(db, make_worker): + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + # The entry's timing changed: its identity changes, so the overdue slot + # on the old ledger row never fires. + hourly = Schedule.from_cron("0 * * * *") + changed = make_worker([(_TickJob(), hourly)]) + changed.maybe_schedule_jobs() + + assert not JobRequest.query.exists() + assert ScheduleState.query.count() == 2 + + +def test_should_enqueue_false_still_consumes_the_slot(db, make_worker): + job = _NeverEnqueueJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + _run_schedule_pass(worker) + + assert not JobRequest.query.exists() + # The slot was evaluated and skipped — the ledger advanced so it isn't + # re-offered every tick. + state = ScheduleState.query.get(schedule_key=schedule_entry_key(job, EVERY_MINUTE)) + assert timezone.now() - state.last_enqueued_slot < datetime.timedelta(seconds=60) + + +def test_legacy_pre_enqueued_row_for_the_same_slot_dedupes(db, make_worker): + """Upgrade path: a row pre-enqueued by the old scheduler for a slot must + not double-fire when the ledger also reaches that slot.""" + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + # Recreate what the old scheduler left behind: a pending row keyed to the + # slot the ledger will fire next — for an every-minute schedule, the + # latest due slot is the current minute boundary. + due_slot = timezone.localtime().replace(second=0, microsecond=0) + JobRequest.query.create( + job_class=jobs_registry.get_job_class_name(_TickJob), + parameters={"args": [], "kwargs": {}}, + queue="default", + concurrency_key=scheduled_concurrency_key(job, due_slot), + start_at=due_slot, + ) + + _run_schedule_pass(worker) + + assert JobRequest.query.count() == 1 + # The ledger still advanced past the deduped slot. + state = ScheduleState.query.get(schedule_key=schedule_entry_key(job, EVERY_MINUTE)) + assert timezone.localtime(state.last_enqueued_slot) == due_slot + + +def test_completed_run_for_the_slot_blocks_reenqueue(db, make_worker): + """Rolling-upgrade path: a slot whose legacy pre-enqueued row already ran + to completion (only a JobResult remains) must not fire a second time.""" + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + due_slot = timezone.localtime().replace(second=0, microsecond=0) + JobResult.query.create( + job_process_uuid=uuid.uuid4(), + job_request_uuid=uuid.uuid4(), + job_class=jobs_registry.get_job_class_name(_TickJob), + status=JobResultStatuses.SUCCESSFUL, + concurrency_key=scheduled_concurrency_key(job, due_slot), + ) + + _run_schedule_pass(worker) + + assert not JobRequest.query.exists() + # The ledger still advanced — the slot is accounted for, just not re-run. + state = ScheduleState.query.get(schedule_key=schedule_entry_key(job, EVERY_MINUTE)) + assert timezone.localtime(state.last_enqueued_slot) == due_slot + + +def test_duplicate_schedule_entries_are_rejected(db): + with pytest.raises(ValueError, match="Duplicate JOBS_SCHEDULE entry"): + load_schedule([(_TickJob(), "* * * * *"), (_TickJob(), "* * * * *")]) + + +def test_long_command_schedules_and_fires(db, make_worker): + """A long shell command must survive the whole path: ledger init (the + unbounded schedule_key) AND the due-slot enqueue (the 255-char + concurrency_key bound on JobRequest).""" + command = "echo " + "x" * 300 + job = ScheduledCommand(command) + worker = make_worker([(job, EVERY_MINUTE)]) + + worker.maybe_schedule_jobs() + assert ScheduleState.query.count() == 1 + + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + _run_schedule_pass(worker) + + job_request = JobRequest.query.get() + assert len(job_request.concurrency_key) <= 255 + + +def test_broken_entry_does_not_starve_later_entries(db, make_worker): + """A failing entry is contained per-entry: the entries after it still + get evaluated in the same pass.""" + broken, healthy = _RaisingEnqueueJob(), _TickJob() + worker = make_worker([(broken, EVERY_MINUTE), (healthy, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(broken, EVERY_MINUTE, minutes=2) + _backdate_ledger(healthy, EVERY_MINUTE, minutes=2) + + _run_schedule_pass(worker) + + job_request = JobRequest.query.get() + assert job_request.job_class == jobs_registry.get_job_class_name(_TickJob) + # The broken entry's claim rolled back with the failed enqueue, so its + # slot stays available to retry next pass. + broken_state = ScheduleState.query.get( + schedule_key=schedule_entry_key(broken, EVERY_MINUTE) + ) + assert timezone.now() - broken_state.last_enqueued_slot > datetime.timedelta( + seconds=90 + ) + + +def test_completed_older_slot_does_not_block_later_slots(db, make_worker): + """The completed-run guard is slot-specific: a JobResult for an earlier + slot must not suppress the current slot's fire.""" + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=5) + + older_slot = (timezone.localtime() - datetime.timedelta(minutes=3)).replace( + second=0, microsecond=0 + ) + JobResult.query.create( + job_process_uuid=uuid.uuid4(), + job_request_uuid=uuid.uuid4(), + job_class=jobs_registry.get_job_class_name(_TickJob), + status=JobResultStatuses.SUCCESSFUL, + concurrency_key=scheduled_concurrency_key(job, older_slot), + ) + + _run_schedule_pass(worker) + + assert JobRequest.query.exists() + + +def test_stale_claim_enqueues_nothing(db, make_worker): + """The optimistic claim: a pass working from a stale ledger read (another + worker advanced the row mid-flight) must not enqueue.""" + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + stale_slot = ScheduleState.query.get( + schedule_key=schedule_entry_key(job, EVERY_MINUTE) + ).last_enqueued_slot + + _run_schedule_pass(worker) # advances the ledger and fires the slot + assert JobRequest.query.count() == 1 + + # Replay the claim with the stale pre-advance value. + worker._schedule_entry( + job, + EVERY_MINUTE, + schedule_entry_key(job, EVERY_MINUTE), + stale_slot, + timezone.localtime(), + ) + + assert JobRequest.query.count() == 1 + + +def test_schedule_gate_blocks_back_to_back_passes(db, make_worker): + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + _run_schedule_pass(worker) + assert JobRequest.query.count() == 1 + + # Another due slot exists, but the gate was stamped at the start of the + # pass above — an immediate second call does nothing. + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + worker.maybe_schedule_jobs() + + assert JobRequest.query.count() == 1 + + +def test_missed_slots_older_than_catchup_window_are_skipped(db, make_worker): + """A stale ledger row — e.g. an entry removed from JOBS_SCHEDULE and + re-added much later — advances without firing a long-gone slot.""" + yearly = Schedule.from_cron("@yearly") + job = _TickJob() + worker = make_worker([(job, yearly)]) + worker.maybe_schedule_jobs() + ScheduleState.query.filter(schedule_key=schedule_entry_key(job, yearly)).update( + last_enqueued_slot=timezone.now() - datetime.timedelta(days=730) + ) + + _run_schedule_pass(worker) + + assert not JobRequest.query.exists() + state = ScheduleState.query.get(schedule_key=schedule_entry_key(job, yearly)) + assert timezone.now() - state.last_enqueued_slot < datetime.timedelta(seconds=60) + + +def test_catchup_window_is_floored(db, make_worker, settings): + """A tiny window must not classify ordinary between-pass slots as stale + and disable scheduling.""" + settings.JOBS_SCHEDULE_CATCHUP_WINDOW = 0 + job = _TickJob() + worker = make_worker([(job, EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + _backdate_ledger(job, EVERY_MINUTE, minutes=2) + + _run_schedule_pass(worker) + + assert JobRequest.query.exists() + + +def test_long_commands_get_distinct_keys(db): + """The digest is what keeps two long commands with a shared prefix from + colliding on one ledger row.""" + prefix = "echo " + "x" * 250 + first = ScheduledCommand(prefix + "AAA") + second = ScheduledCommand(prefix + "BBB") + + assert first.default_concurrency_key() != second.default_concurrency_key() + + +def test_command_that_fits_keeps_its_raw_key(db): + """Commands whose slot-stamped key fits the 255-char bound keep the raw + command as their key — matching what pre-ledger workers produced, so + slot dedupe holds across the upgrade.""" + command = "x" * 234 + job = ScheduledCommand(command) + + assert job.default_concurrency_key() == command + slot = timezone.localtime().replace(second=0, microsecond=0) + assert len(scheduled_concurrency_key(job, slot)) <= 255 + + +def test_legacy_sweep_runs_with_an_empty_schedule(db, make_worker): + """Pre-ledger transition: a release that removes the last schedule entry + must still sweep rows an old worker re-created.""" + future_slot = (timezone.localtime() + datetime.timedelta(hours=2)).replace( + second=0, microsecond=0 + ) + JobRequest.query.create( + job_class="app.gone.RemovedJob", + parameters={"args": [], "kwargs": {}}, + queue="default", + concurrency_key=f":scheduled:{int(future_slot.timestamp())}", + start_at=future_slot, + ) + + worker = make_worker([]) + worker.maybe_schedule_jobs() + + assert not JobRequest.query.exists() + + +def test_legacy_future_rows_are_swept_with_provenance(db, make_worker): + """Pre-ledger transition: a future row an old worker re-created after + the sweep migration is deleted by the schedule pass — but only when its + key's slot stamp matches its own start time (true provenance). A user + row that merely looks similar survives.""" + future_slot = (timezone.localtime() + datetime.timedelta(hours=2)).replace( + second=0, microsecond=0 + ) + legacy = JobRequest.query.create( + job_class=jobs_registry.get_job_class_name(_TickJob), + parameters={"args": [], "kwargs": {}}, + queue="default", + concurrency_key=f":scheduled:{int(future_slot.timestamp())}", + start_at=future_slot, + ) + lookalike = JobRequest.query.create( + job_class=jobs_registry.get_job_class_name(_TickJob), + parameters={"args": [], "kwargs": {}}, + queue="default", + concurrency_key="batch:scheduled:123", + start_at=future_slot, + ) + + worker = make_worker([(_TickJob(), EVERY_MINUTE)]) + worker.maybe_schedule_jobs() + + remaining = set(JobRequest.query.all().values_list("id", flat=True)) + assert legacy.id not in remaining + assert lookalike.id in remaining + + +def test_day_combination_flag_is_part_of_schedule_identity(db): + """Two keyword schedules differing only in combine_days_with_or match + different slot sets — they must not collide on one ledger key.""" + and_days = Schedule(day_of_month=15, day_of_week=5) + or_days = Schedule(day_of_month=15, day_of_week=5, combine_days_with_or=True) + job = _TickJob() + + assert schedule_entry_key(job, and_days) != schedule_entry_key(job, or_days) + # And configuring both is legal, not a false duplicate. + load_schedule([(_TickJob(), and_days), (_TickJob(), or_days)]) + + +def test_prune_chore_deletes_only_stale_ledger_rows(db, settings): + from plain.jobs.chores import PruneScheduleLedger + + settings.JOBS_SCHEDULE = [(jobs_registry.get_job_class_name(_TickJob), "* * * * *")] + current_key = schedule_entry_key(_TickJob(), EVERY_MINUTE) + ScheduleState.query.create( + schedule_key=current_key, last_enqueued_slot=timezone.now() + ) + ScheduleState.query.create( + schedule_key="app.gone.RemovedJob::@daily", + last_enqueued_slot=timezone.now(), + ) + + PruneScheduleLedger().run() + + assert list(ScheduleState.query.all().values_list("schedule_key", flat=True)) == [ + current_key + ] + + +def test_pending_job_for_removed_class_errors_clearly(db): + """The backstop: a row whose class is gone produces an errored result + naming JobClassNotRegistered, not a bare KeyError.""" + job_process = JobProcess.query.create( + job_request_uuid=uuid.uuid4(), + job_class="app.gone.RemovedJob", + parameters={"args": [], "kwargs": {}}, + worker_id=uuid.uuid4(), + ) + + result = job_process.run() + + assert result.status == JobResultStatuses.ERRORED + assert "JobClassNotRegistered" in result.error