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