Debug a Python job scheduler

0
Sign in
Sign in

You own a small in-process job scheduler. It runs on a fake clock so it can be tested. Users report that:

  • a nightly report scheduled for exactly the end of a window sometimes waits a whole extra cycle,
  • urgent jobs run after routine ones,
  • an every-10-minutes job runs once and then goes quiet,
  • a job configured for 2 retries only retries once,
  • and after someone cancels a job and schedules it again under the same name, it never runs.

Find and fix the bugs. For each one: reproduce it with a failing test, narrow it to the line, fix it, and keep the test. Then say what you would change long-term so this class of bug does not come back. Do not rewrite the scheduler from scratch. You have about 45 minutes and may run code.

Spec

  • schedule(name, fn, run_at, priority=0, interval=None, max_retries=0) adds a job. Names are unique: scheduling a name that is already scheduled replaces the old job.
  • run_until(t) runs every job with run_at <= t, in order of run_at, then higher priority first, then the order they were scheduled (a retry or the next run of a recurring job counts as scheduled when it is queued again). While a job runs, now is that job's time; afterwards now is t.
  • A recurring job's next run is its previous planned time plus interval, so retries and late runs never shift its cadence.
  • A job that raises is retried up to max_retries times; the k-th retry runs 2 ** (k - 1) ticks after the failure. When the retries run out, history records "failed", the name goes into failed, and a recurring job stops.
  • cancel(name) returns True if the job was scheduled, False otherwise.
  • history is a list of (time, name, outcome) with outcome "ok", "retry" or "failed".
"""A small in-process job scheduler driven by a fake clock."""
from __future__ import annotations

import heapq
from dataclasses import dataclass, field
from typing import Callable, Optional


@dataclass(order=True)
class Job:
    run_at: int
    priority: int
    name: str = field(compare=False)
    fn: Callable[[], object] = field(compare=False)
    interval: Optional[int] = field(default=None, compare=False)
    max_retries: int = field(default=0, compare=False)
    attempts: int = field(default=0, compare=False)


class Scheduler:
    def __init__(self) -> None:
        self.now = 0
        self._heap: list = []
        self._cancelled: set = set()
        self.history: list = []  # (time, name, "ok" | "retry" | "failed")
        self.failed: list = []

    def schedule(self, name, fn, run_at, priority=0, interval=None, max_retries=0):
        heapq.heappush(self._heap, Job(run_at, priority, name, fn, interval, max_retries))

    def cancel(self, name) -> bool:
        self._cancelled.add(name)
        return True

    def run_until(self, t) -> None:
        self.now = t
        while self._heap and self._heap[0].run_at < t:
            job = heapq.heappop(self._heap)
            if job.name in self._cancelled:
                continue
            try:
                job.fn()
            except Exception:
                job.attempts += 1
                if job.attempts >= job.max_retries:
                    self.history.append((job.run_at, job.name, "failed"))
                    self.failed.append(job.name)
                else:
                    self.history.append((job.run_at, job.name, "retry"))
                    job.run_at = job.run_at + 2 ** (job.attempts - 1)
                    heapq.heappush(self._heap, job)
                continue
            self.history.append((job.run_at, job.name, "ok"))
            job.attempts = 0
            if job.interval:
                job.run_at = self.now + job.interval
                heapq.heappush(self._heap, job)

These tests come with the code. All four fail today.

from scheduler import Scheduler


def ok():
    return None


def test_job_due_exactly_at_t_runs():
    s = Scheduler()
    s.schedule("report", ok, run_at=10)
    s.run_until(10)
    assert s.history == [(10, "report", "ok")]


def test_higher_priority_runs_first():
    s = Scheduler()
    s.schedule("low", ok, run_at=5, priority=1)
    s.schedule("high", ok, run_at=5, priority=9)
    s.run_until(5)
    assert [name for _, name, _ in s.history] == ["high", "low"]


def test_recurring_job_keeps_its_cadence():
    s = Scheduler()
    s.schedule("tick", ok, run_at=0, interval=10)
    s.run_until(35)
    assert [t for t, _, _ in s.history] == [0, 10, 20, 30]


def test_max_retries_means_retries_after_the_first_try():
    calls = []

    def flaky():
        calls.append(1)
        raise RuntimeError("boom")

    s = Scheduler()
    s.schedule("flaky", flaky, run_at=0, max_retries=2)
    s.run_until(100)
    assert len(calls) == 3
    assert s.history == [(0, "flaky", "retry"), (1, "flaky", "retry"), (3, "flaky", "failed")]
    assert s.failed == ["flaky"]
OpenAIPractical codingMedium

Sign in to see the worked answer

The full answer, chapter by chapter, with what interviewers expect at each level. Free with an account: one full answer per company.

Sign in to read it