Debug a Python job scheduler: Frontier Interview's solution

Question: Debug a Python job scheduler

Companies: OpenAI

Difficulty: Medium

Problem statement

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

How to run the round

This round grades how you debug, not how fast you rewrite. The interviewer watches for a loop you repeat for every symptom:

Reproduce the symptom with the smallest possible call sequence, as a failing test.
Narrow it to one line by reading the code path that test exercises, printing state if reading is not enough.
Fix that one line, and say why the fix is right, not just that the test turned green.
Keep the test as a regression test.
Name the long-term fix: the change to the design that stops this class of bug, which you might not make today.

Say the loop out loud once at the start, then use it. Do not rewrite the scheduler from scratch: the interviewer wants to see you find the planted bugs, and a rewrite hides both the finds and the misses. A reasonable 45-minute budget:

Minutes	What
0–5	Read `scheduler.py` top to bottom. List what each field means. Run the given tests; four fail.
5–25	One failing test at a time: reproduce, narrow, fix, rerun everything.
25–35	Read the code again for bugs no test covers yet (cancel, ties, replacement). Write a test first, then fix.
35–45	Long-term fixes and follow-ups: validation, misfire policy, threads, persistence.

Read the code first

Five minutes of reading pays for itself. Write down what the data structures promise before you look for bugs:

_heap is ordered by the Job dataclass comparison. order=True compares fields in declaration order, skipping compare=False ones, so the key is (run_at, priority). That already raises two questions: lower priority pops first, which is the opposite of the spec, and nothing breaks ties.
_cancelled is a set of names that is only ever added to.
run_until sets self.now = t before it runs anything, so every self.now inside the loop is the end of the window, not the time of the job being run.

Three of the six bugs are visible from this list before you run a single test. Say them, but still prove each one with a test before you change it.

Reproduce and narrow each failure

test_job_due_exactly_at_t_runs fails. History is empty. The loop condition is self._heap[0].run_at < t. The spec says every job with run_at <= t runs. Off by one at the boundary.

test_higher_priority_runs_first fails. low runs before high. The heap is a min-heap over (run_at, priority), so priority 1 pops before priority 9. Negate the priority in the key.

test_recurring_job_keeps_its_cadence fails. After fixing the boundary, history is [0] only. Print job.run_at after the reschedule: it is 45, not 10. The line is job.run_at = self.now + job.interval and self.now is 35, the end of the window. Even with the clock moved to the job's time, computing the next slot from "when it ran" makes a retried job drift: a job planned for 0 that succeeds on a retry at 1 would next run at 11, 21, 31. The next slot must come from the planned slot: job.run_at += job.interval.

test_max_retries_means_retries_after_the_first_try fails. Only two calls, not three. attempts counts failures and is incremented before the check, so with max_retries=2 the second failure hits attempts >= 2 and gives up. The spec says max_retries is the number of retries after the first try, so the condition is attempts > max_retries. Check the backoff while you are there: the k-th retry waits 2 ** (k - 1) ticks, so 1 then 2, which the test pins as times 1 and 3.

Planted bugs

#	Where	Symptom	Fix
1	`while ... run_at < t`	A job due exactly at `t` waits for the next call	`<=`
2	`Job(order=True)` key	Priority 1 runs before priority 9	Key on `-priority`
3	No tie-breaker in the key	Jobs with equal time and priority run in an arbitrary order (`heapq` is not stable)	Add an increasing sequence number to the key
4	`job.run_at = self.now + interval` with `self.now = t`	A recurring job runs once per `run_until` call and drifts after retries	`job.run_at += interval`; set `self.now` to each job's time while it runs
5	`attempts >= max_retries`	One retry fewer than asked	`attempts > max_retries`
6	`_cancelled` set of names, never cleared	Cancel then schedule the same name again: the new job never runs. Scheduling a name twice runs both. `cancel` always returns `True`	A token per `schedule()` call; the heap entry runs only if its token is the live one

Bugs 3 and 6 have no given test. Finding them is what separates a pass from a strong pass: they come from reading the code against the spec, not from the red tests.

Bug 3 needs a careful test. Two equal jobs often pop in the right order by luck. Five jobs with the same (run_at, priority) reliably come out of heapq in a different order than they went in, and the dataclass comparison never looks at the name.

Bug 6 needs two tests: cancel followed by a reschedule of the same name, and a schedule of a name that is already scheduled.

The fix

"""A small in-process job scheduler driven by a fake clock."""
from __future__ import annotations

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


@dataclass
class Job:
    name: str
    fn: Callable[[], object]
    run_at: int
    priority: int = 0
    interval: Optional[int] = None
    max_retries: int = 0
    attempts: int = 0  # failures of the current occurrence
    token: int = 0     # which schedule() call this job belongs to


class Scheduler:
    def __init__(self) -> None:
        self.now = 0
        # (run_at, -priority, seq, token, job): seq keeps equal keys FIFO and
        # means the heap never has to compare two Job objects.
        self._heap: list = []
        self._seq = itertools.count()
        self._live: dict = {}  # name -> token of the job that may still run
        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) -> None:
        if interval is not None and interval <= 0:
            raise ValueError("interval must be a positive number of ticks")
        if max_retries < 0:
            raise ValueError("max_retries must be >= 0")
        token = next(self._seq)
        self._live[name] = token  # replaces any earlier job with this name
        self._push(Job(name, fn, run_at, priority, interval, max_retries, token=token))

    def cancel(self, name) -> bool:
        return self._live.pop(name, None) is not None

    def run_until(self, t) -> None:
        while self._heap and self._heap[0][0] <= t:
            run_at, _, _, token, job = heapq.heappop(self._heap)
            if self._live.get(job.name) != token:
                continue  # cancelled or replaced: a stale heap entry
            self.now = run_at
            try:
                job.fn()
            except Exception:
                self._on_failure(job)
                continue
            self.history.append((run_at, job.name, "ok"))
            job.attempts = 0
            if job.interval is not None:
                # Next slot comes from the slot we planned, not from when the
                # job happened to run, so a late or retried run never drifts.
                job.run_at += job.interval
                self._push(job)
            else:
                self._retire(job)
        self.now = max(self.now, t)

    def _on_failure(self, job: Job) -> None:
        job.attempts += 1
        if job.attempts > job.max_retries:
            self.history.append((self.now, job.name, "failed"))
            self.failed.append(job.name)
            self._retire(job)
            return
        self.history.append((self.now, job.name, "retry"))
        retry_at = self.now + 2 ** (job.attempts - 1)
        heapq.heappush(self._heap, (retry_at, -job.priority, next(self._seq), job.token, job))

    def _retire(self, job: Job) -> None:
        # Forget the name only if the job did not reschedule it while running.
        if self._live.get(job.name) == job.token:
            del self._live[job.name]

    def _push(self, job: Job) -> None:
        heapq.heappush(self._heap, (job.run_at, -job.priority, next(self._seq), job.token, job))

Why each change is right, in one line each:

Tuple key with a sequence number. heapq compares tuples left to right. seq is unique, so two entries never tie, the order is FIFO among equals, and Python never falls through to comparing Job objects (which would raise TypeError once order=True is gone).
Tokens instead of a cancelled set. This is lazy deletion. Removing an arbitrary entry from a heap costs O(n); marking it dead costs O(1) and the pop skips it. A token per schedule() call means "cancelled" and "replaced" are the same check, and a later schedule of the same name is never confused with the old one.
_retire checks the token. A job may call cancel or schedule on its own name while it runs. Deleting the name blindly would either raise KeyError or wipe the job it just scheduled.
self.now follows the job. Jobs read self.now, and retries are timed from it. It ends at t, never before the last job run.

Regression tests

Every bug gets a test that failed before its fix. The four given tests cover bugs 1, 2, 4 and 5. These cover the bugs no test caught, and the corners the fix introduced.

from scheduler import Scheduler


def ok():
    return None


def test_equal_time_and_priority_run_in_schedule_order():
    s = Scheduler()
    for name in ["a", "b", "c", "d", "e"]:
        s.schedule(name, ok, run_at=7)
    s.run_until(7)
    assert [name for _, name, _ in s.history] == ["a", "b", "c", "d", "e"]


def test_cancel_then_reschedule_runs_the_new_job():
    s = Scheduler()
    s.schedule("sync", ok, run_at=5)
    assert s.cancel("sync") is True
    s.schedule("sync", ok, run_at=8)
    s.run_until(10)
    assert s.history == [(8, "sync", "ok")]


def test_cancel_unknown_name_returns_false():
    s = Scheduler()
    assert s.cancel("nope") is False


def test_scheduling_a_name_again_replaces_it():
    s = Scheduler()
    s.schedule("sync", ok, run_at=5)
    s.schedule("sync", ok, run_at=9)
    s.run_until(10)
    assert s.history == [(9, "sync", "ok")]


def test_cancel_stops_a_recurring_job():
    s = Scheduler()
    s.schedule("tick", ok, run_at=0, interval=10)
    s.run_until(15)
    s.cancel("tick")
    s.run_until(100)
    assert [t for t, _, _ in s.history] == [0, 10]

from scheduler import Scheduler


def ok():
    return None


def test_zero_interval_is_rejected():
    s = Scheduler()
    try:
        s.schedule("spin", ok, run_at=0, interval=0)
    except ValueError:
        return
    assert False, "interval=0 would loop forever inside run_until"


def test_retried_recurring_job_keeps_its_cadence():
    state = {"n": 0}

    def fails_once():
        state["n"] += 1
        if state["n"] == 1:
            raise RuntimeError("cold cache")

    s = Scheduler()
    s.schedule("tick", fails_once, run_at=0, interval=10, max_retries=1)
    s.run_until(25)
    assert s.history == [(0, "tick", "retry"), (1, "tick", "ok"), (10, "tick", "ok"), (20, "tick", "ok")]


def test_job_that_cancels_itself_does_not_crash():
    s = Scheduler()
    s.schedule("once", lambda: s.cancel("once"), run_at=1)
    s.run_until(5)
    assert s.history == [(1, "once", "ok")]


def test_job_that_reschedules_itself_keeps_the_new_run():
    s = Scheduler()
    s.schedule("chain", lambda: s.schedule("chain", ok, run_at=s.now + 3), run_at=1)
    s.run_until(10)
    assert s.history == [(1, "chain", "ok"), (4, "chain", "ok")]


def test_clock_ends_at_t():
    s = Scheduler()
    s.schedule("a", ok, run_at=2)
    s.run_until(9)
    assert s.now == 9

Long-term fixes

The one-line fixes make the tests pass. A senior answer also says what would stop the next bug of the same kind:

One ordering function, not an implicit dataclass order. Bugs 2 and 3 both came from letting @dataclass(order=True) decide the heap order. An explicit _key(job) that returns (run_at, -priority, seq) is one place to read and test.
Validate at the edge. interval=0 or a negative interval turns run_until into an infinite loop. max_retries=-1 silently means "never retry". Reject both in schedule with a clear error.
Name the misfire policy. If a retry runs past the next planned slot, the fixed code runs the missed slot right away (catch up). Some jobs want that (billing); others want "skip to the next future slot" (a cache warmer). Make it a per-job option instead of an accident.
Separate "what time is it" from "what is due". A real scheduler reads a monotonic clock. Injecting the clock (as this fake clock does) is what made every bug here reproducible in a unit test; keep that seam.
Property test. Generate random schedules, cancels and failures, and check invariants after every run_until(t): the history is sorted by time, no cancelled token ever runs, no job due at or before t is left in the heap, and a recurring job's ok-times are its start plus multiples of its interval.

Follow-ups

"Jobs take real time and the scheduler is shared by threads." Hold a lock while you pop and push, but never while fn() runs, or one slow job blocks every other. Run jobs on a worker pool. Cancellation must then be checked when the job is popped and again before a recurrence is pushed, because cancel can land while the job is running. That is exactly what the token check already does.
"The process restarts. Nothing may be lost." Persist (name, token, run_at, attempts) before a job runs, and mark it done after. On restart, a job that was started but not marked done ran zero or one times, so it runs again: at-least-once. Jobs must be idempotent, or carry an idempotency key built from (name, planned run_at).
"Ten thousand recurring jobs fire at midnight." The heap handles it (O(log n) per pop), but the downstream does not. Add jitter to recurring jobs, or a concurrency limit per job class.
"How would you find bug 3 in production?" You usually do not; it shows up as "this report sometimes runs before its input is ready". That is the argument for the property test above, and for making dependencies explicit instead of relying on schedule order.

Interviewer expectations

Mid-level (L4)

Reproduces each given failure with a test, finds the line, and fixes bugs 1, 2, 4 and 5 without breaking the others.
Reruns the whole test file after every fix, not only the test being worked on.
Explains each fix in terms of the spec ("max_retries counts retries, not tries").

Senior (L5)

Reads the code before running it and calls out the dataclass ordering and the never-cleared cancelled set early.
Finds bug 3 or bug 6 without a failing test, writes the test first, and fixes it with a sequence number and tokens (lazy deletion).
Notices the drift problem goes beyond "uses t": the next slot must come from the planned slot, or retries shift the cadence.

Staff (L6+)

Finds all six, plus the infinite loop on interval <= 0 and the self-cancel KeyError the obvious fix for bug 6 would introduce.
Proposes the structural fixes (explicit ordering key, validation, a named misfire policy, property tests) and says which ones they would ship now.
Handles the threading and restart follow-ups crisply: no lock across fn(), at-least-once with idempotent jobs.