Delayed and recurring task scheduler

Q: How does a real background thread drive this?

A loop waits on a condition variable with a timeout of next_run_in() (or indefinitely when there is nothing to run), then calls run_pending(). schedule and cancel notify the condition, so a newly added task that is due sooner than the current wait wakes the loop early. Jobs themselves should be handed to a worker pool, so one slow job does not delay everything behind it.

Q: How would this look in Go, and in production?

In Go the scheduler goroutine uses container/heap and a select over a time.Timer reset to the next due time, a channel of new tasks, a cancel channel and ctx.Done(). Owning the heap in one goroutine means no mutex is needed. In production you also decide what happens across restarts. An in-memory schedule is lost, so durable jobs live in a database or in a system like Kubernetes CronJobs, and this in-process scheduler is for short-lived housekeeping.

mediumScheduling and queues

Build the core of an in-process job scheduler, the part of an agent that runs "rotate logs in 2 seconds" or "probe the backend every 3 seconds". The scheduler is tick-driven: something outside it calls run_pending() regularly, and it runs whatever is due. That keeps it deterministic to test, and the FAQ covers how a real background loop drives it.

API (Java: schedule(delay, fn) / schedule(delay, fn, Double interval), runPending(), nextRunIn() returning Double or null, Runnable jobs)

◈ DIAGRAM

Scheduler(clock=time.monotonic)
schedule(delay: float, fn, interval=None) -> int   # task id, starting at 1
cancel(task_id: int) -> bool                       # False if unknown or already finished
run_pending() -> int                               # number of tasks run in this call
next_run_in() -> float or None                     # seconds until the next task is due
errors: list[str]                                  # "task <id>: <message>" per failed run

Rules

A task scheduled at time t with delay d is due at t + d. run_pending() runs every task due at or before now, earliest due time first. Tasks with the same due time run in the order they were scheduled.
A one-shot task is removed after it runs. A recurring task (with interval) is rescheduled for due + interval. If that time has already passed, because run_pending was called late, it goes to now + interval instead: missed runs are skipped, not replayed in a burst. So a recurring task runs at most once per run_pending call.
If a job raises, record "task <id>: <message>" in errors and keep going. A recurring job that fails is still rescheduled.
A job may cancel itself or other tasks while it runs.
Cancelling must not require scanning all tasks.
delay < 0 or interval <= 0 raises ValueError.

job(name) in the examples appends name to a log list. Outputs are the return values of each call in order, then the final list or lists.

Examples

Example 1

Input: t=0 s.schedule(5, job("backup")) t=0 s.schedule(2, job("rotate")) t=0 s.schedule(3, job("probe"), interval=3) t=0 s.schedule(4, job("noop")) t=0 s.cancel(4) t=1 s.run_pending() t=2 s.run_pending() t=6 s.run_pending() t=9 s.run_pending() t=9 s.next_run_in() t=9 s.cancel(3) t=20 s.run_pending() t=20 s.cancel(3) t=20 s.next_run_in() log

Output: 1 2 3 4 true 0 1 2 1 3.0 true 0 false null [rotate, probe, backup, probe]

Explanation: At t=6, probe (due 3) and backup (due 5) run in due order. Probe's next slot, 6, is not in the future, so it moves to 9 rather than running twice in one call.

Example 2

Input: t=0 s.schedule(1, boom) # raises OSError("disk full") t=0 s.schedule(1, job("after-boom")) t=0 s.schedule(1, once_then_stop, interval=1) # logs "tick", then cancels itself t=1 s.run_pending() t=5 s.run_pending() s.errors log

Output: 1 2 3 3 0 [task 1: disk full] [after-boom, tick]

Explanation: A failing job is recorded and the rest still run. The recurring job cancels itself on its first run, so nothing is left at t=5.

Hints

Approach

Optimal

Use a min-heap of (due, seq, id) and a dict id -> task. The task stores its current due and seq.

schedule: allocate an id, store the task, push (now + delay, seq, id). seq is a global counter, so equal due times pop in FIFO order.
cancel: tasks.pop(id). O(1). The heap entry stays and is discarded when it reaches the top, because its id is no longer in tasks.
run_pending: first drop stale entries from the top. Then, while the top is due, pop it, run it inside try/except, and either delete the task (one-shot), re-push it with the next due time (recurring), or leave it deleted if the job cancelled itself.
next_run_in: after dropping stale entries, the top of the heap is the answer.

Reading now once at the start of run_pending matters: a slow job cannot make the loop chase a moving target and run tasks that became due only while it was busy.

ComplexityTime O(log n) per schedule and per task run; O(1) cancelSpace O(n) plus stale heap entries

Python

import heapq
import itertools
import time
 
 
class _Task:
    __slots__ = ("id", "fn", "interval", "due", "seq")
 
    def __init__(self, task_id, fn, interval, due, seq):
        self.id, self.fn, self.interval, self.due, self.seq = task_id, fn, interval, due, seq
 
 
class Scheduler:
    def __init__(self, clock=time.monotonic):
        self.clock = clock
        self.tasks = {}                    # id -> live task
        self.heap = []                     # (due, seq, id); stale entries skipped lazily
        self.ids = itertools.count(1)
        self.seqs = itertools.count()
        self.errors = []
 
    def _push(self, task, due):
        task.due, task.seq = due, next(self.seqs)
        heapq.heappush(self.heap, (due, task.seq, task.id))
 
    def schedule(self, delay, fn, interval=None):
        if delay < 0 or (interval is not None and interval <= 0):
            raise ValueError("delay must be >= 0 and interval > 0")
        task = _Task(next(self.ids), fn, interval, 0, 0)
        self.tasks[task.id] = task
        self._push(task, self.clock() + delay)
        return task.id
 
    def cancel(self, task_id):
        return self.tasks.pop(task_id, None) is not None
 
    def _discard_stale(self):
        while self.heap:
            due, seq, tid = self.heap[0]
            task = self.tasks.get(tid)
            if task is not None and task.seq == seq:
                return
            heapq.heappop(self.heap)
 
    def next_run_in(self):
        self._discard_stale()
        if not self.heap:
            return None
        return float(max(0, self.heap[0][0] - self.clock()))
 
    def run_pending(self):
        now = self.clock()
        ran = 0
        self._discard_stale()
        while self.heap and self.heap[0][0] <= now:
            due, seq, tid = heapq.heappop(self.heap)
            task = self.tasks[tid]
            try:
                task.fn()
            except Exception as exc:           # one bad job must not stop the others
                self.errors.append(f"task {tid}: {exc}")
            ran += 1
            if task.interval is None:
                self.tasks.pop(tid, None)
            elif tid in self.tasks:            # the job may have cancelled itself
                nxt = due + task.interval
                if nxt <= now:                 # fell behind: skip missed runs, no burst
                    nxt = now + task.interval
                self._push(task, nxt)
            self._discard_stale()
        return ran

Follow-up questions

Run jobs on a worker pool. How do you stop a slow recurring job from overlapping with its own next run?
Add cron-style schedules using the cron parser's next_after.
Persist the schedule so tasks survive a restart. What do you do about runs missed while the process was down?

Frequently asked questions

A loop waits on a condition variable with a timeout of next_run_in() (or indefinitely when there is nothing to run), then calls run_pending(). schedule and cancel notify the condition, so a newly added task that is due sooner than the current wait wakes the loop early. Jobs themselves should be handed to a worker pool, so one slow job does not delay everything behind it.

In Go the scheduler goroutine uses container/heap and a select over a time.Timer reset to the next due time, a channel of new tasks, a cancel channel and ctx.Done(). Owning the heap in one goroutine means no mutex is needed. In production you also decide what happens across restarts. An in-memory schedule is lost, so durable jobs live in a database or in a system like Kubernetes CronJobs, and this in-process scheduler is for short-lived housekeeping.

This design is fixed-rate: the next run is based on the previous due time, so a probe stays on a steady 3-second grid. Fixed-delay schedules the next run interval after the job finishes, so a slow job pushes every later run back. Fixed-rate suits metrics scraping; fixed-delay suits jobs that must never overlap with themselves.

Problem statement