Replace pre-enqueued scheduled jobs with a ScheduleState ledger

.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.

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

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`              |

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"]

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):

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.

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"

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

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.
 

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")'
+        ),
+    ]

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 = <old>` 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.