The harness, in code · Build it

A tiny DAG scheduler

Implement the scheduler that runs a harness DAG in the same topological order every time, retries each step to its own limit, and names the resume point.

codehardExercise · 3 min · 72 of 77

On this page

A resume point is only worth recording if the next run means the same thing by it. The scheduler that produces one has to order the steps identically every time, find a cycle before it spends anything, and stop where a person can pick up.

The situation

A content pipeline runs nightly as a seven-step harness DAG: fetch (code), then extract, outline and draft (model calls), check (code, links and figures against the source), images (a model call from the outline) and publish (code). The first scheduler had three faults, all found in the week step 5 started failing. It sorted the DAG with a recursive depth-first search, which hit Python's default limit of 1,000 frames on a 1,200-step backfill that chains one step per archived article. It walked the ready steps as a set of names, and because string hashing changes between processes, two runs could order draft and images differently, so "resume at step 5" named a different step after a restart. And it retried nothing, so one timeout in extract reran the night.

The team wants the scheduler behind resume after step 5: one order, per-step retries, and the step where the run stopped. Whether the cause lies upstream of that step stays the harness's call.

The question

Implement the scheduler so that every run of the same DAG takes the same order, a cycle is reported before anything runs, and the resume point is the step that used up its attempts.

def solve(steps, failures):
    """steps: [(name, deps, retries), ...]; deps are the names this step waits
    for, each declared somewhere in steps; retries is the attempts allowed
    after the first. failures: name -> how many attempts of that step fail
    before one succeeds; a name not in failures succeeds first time.
    Runs one step at a time in topological order, ties broken by the order
    the steps are declared, and stops at the first step that uses up its
    attempts. Returns (status, trace, resume):
      status  "done", "failed", or "cycle" (no order exists, nothing runs)
      trace   [(name, attempts), ...] for every step that ran, in run order
      resume  for "failed", the step a resumed run starts from, skipping
              every step before it in trace; otherwise None"""

PIPELINE = [
    ("fetch", [], 2),
    ("extract", ["fetch"], 1),
    ("outline", ["extract"], 1),
    ("draft", ["outline"], 2),
    ("check", ["draft"], 1),
    ("images", ["outline"], 1),
    ("publish", ["check", "images"], 0),
]

solve(PIPELINE, {})
    -> ("done", [("fetch", 1), ("extract", 1), ("outline", 1), ("draft", 1),
                 ("check", 1), ("images", 1), ("publish", 1)], None)
solve(PIPELINE, {"draft": 2})
    -> ("done", [("fetch", 1), ("extract", 1), ("outline", 1), ("draft", 3),
                 ("check", 1), ("images", 1), ("publish", 1)], None)
solve(PIPELINE, {"check": 2})
    -> ("failed", [("fetch", 1), ("extract", 1), ("outline", 1), ("draft", 1), ("check", 2)], "check")
solve([("a", ["b"], 0), ("b", ["a"], 0)], {})
    -> ("cycle", [], None)

A strong answer · Rubric · Where answers go wrong · Solution

Write your answer first, then open

A strong answer

Order first, run second. Use Kahn's peel from cycles and topological order: count each step's unmet dependencies, put the steps with none in a ready set, take one, release its children. Make the ready set a min-heap on declaration index, so ties break the same way on every run: draft, declared fourth, always runs before images, declared sixth, and "resume at check" survives a restart. The peel is a loop, so a 5,000-step chain costs no stack. If it emits fewer steps than were declared, the rest wait on a cycle: return cycle before anything runs, because a scheduler that finds the cycle halfway has paid for the first half.

Then walk the order. A step with r retries gets r + 1 attempts; with f scripted failures it succeeds on attempt f + 1 when f ≤ r, and the trace records that count, so ("draft", 3) shows the two retries it absorbed. When f > r, record r + 1 attempts and stop with that step as the resume point. Every earlier entry succeeded, so a resumed run skips them: in the third example it skips four steps, three of them model calls, and reruns only check. This is the outer loop of a harness in 200 lines.

publish has a limit of 0, because a step with side effects retries only behind an idempotency key, as idempotency sets out. The scheduler stops at the first failure rather than running images, which does not need check. The evidence that would change that: an independent branch slow enough to be worth finishing before a person looks.

Rubric

computes the whole order before running anything, and reports a cycle with an empty trace
breaks ties by declaration order, so two runs of one DAG take the same order
sorts iteratively, so a chain of thousands of steps does not exhaust the stack
gives each step retries + 1 attempts and records the attempts used in the trace
stops at the first step that exhausts its attempts and returns it as the resume point, with every earlier trace entry a success

Where answers go wrong

"A recursive depth-first sort, reversed post-order." It orders the pipeline correctly and raises RecursionError on the backfill; the 5,000-step chain in the tests is there for this answer.

"Keep the ready steps in a set." The order then depends on string hashing, which differs between processes, so the resume file's step 5 is check on one run and draft on the next, and a resumed run that skips four steps skips images, which never ran.

"On any failure, rerun the DAG from the top." The four steps before check run again, three of them model calls, for a failure in a code step that needed none of them.

Solution

Python 3 · standard library62 lines · 10 of 10 tests passing

"""A tiny DAG scheduler — run a harness DAG in topological order with per-step retries, and name the resume point."""
import heapq


def run_order(steps):
    """Kahn's peel, ties broken by declaration order so every run is the same run. None means a cycle."""
    index = {name: i for i, (name, _, _) in enumerate(steps)}
    waiting = [len(deps) for _, deps, _ in steps]      # unmet dependencies per step
    children = [[] for _ in steps]
    for i, (_, deps, _) in enumerate(steps):
        for d in deps:
            children[index[d]].append(i)
    ready = [i for i, n in enumerate(waiting) if n == 0]
    heapq.heapify(ready)                                # smallest declaration index first
    order = []
    while ready:                                        # iterative: a 5,000-step chain is no deeper than a 3-step one
        i = heapq.heappop(ready)
        order.append(i)
        for c in children[i]:
            waiting[c] -= 1
            if waiting[c] == 0:
                heapq.heappush(ready, c)
    return order if len(order) == len(steps) else None  # a step never released is waiting on a cycle


def solve(steps, failures):
    """steps: [(name, deps, retries), ...]; deps are the names this step waits
    for, each declared somewhere in steps; retries is the attempts allowed
    after the first. failures: name -> how many attempts of that step fail
    before one succeeds; a name not in failures succeeds first time.
    Runs one step at a time in topological order, ties broken by the order
    the steps are declared, and stops at the first step that uses up its
    attempts. Returns (status, trace, resume):
      status  "done", "failed", or "cycle" (no order exists, nothing runs)
      trace   [(name, attempts), ...] for every step that ran, in run order
      resume  for "failed", the step a resumed run starts from, skipping
              every step before it in trace; otherwise None"""
    order = run_order(steps)
    if order is None:
        return ("cycle", [], None)          # validate before spending: nothing has run yet
    trace = []
    for i in order:
        name, _, retries = steps[i]
        fails = failures.get(name, 0)
        if fails > retries:
            trace.append((name, retries + 1))   # every permitted attempt was spent
            return ("failed", trace, name)      # invariant: every step before it in trace succeeded
        trace.append((name, fails + 1))
    return ("done", trace, None)


PIPELINE = [
    ("fetch", [], 2),
    ("extract", ["fetch"], 1),
    ("outline", ["extract"], 1),
    ("draft", ["outline"], 2),
    ("check", ["draft"], 1),
    ("images", ["outline"], 1),
    ("publish", ["check", "images"], 0),
]
ALL_ONCE = [(name, 1) for name, _, _ in PIPELINE]
CHAIN = [(f"s{i}", [f"s{i - 1}"] if i else [], 0) for i in range(5000)]

The cases that ran

TESTS = [
    # the examples on the page
    ((PIPELINE, {}), ("done", ALL_ONCE, None)),
    ((PIPELINE, {"draft": 2}), ("done", ALL_ONCE[:3] + [("draft", 3)] + ALL_ONCE[4:], None)),
    ((PIPELINE, {"check": 2}), ("failed", ALL_ONCE[:4] + [("check", 2)], "check")),
    (([("a", ["b"], 0), ("b", ["a"], 0)], {}), ("cycle", [], None)),
    # a cycle behind a valid first step still runs nothing
    (([("load", [], 0), ("a", ["load", "c"], 0), ("b", ["a"], 0), ("c", ["b"], 0)], {}), ("cycle", [], None)),
    # ties go to declaration order; a step declared before its dependency still waits for it
    (([("summarise", [], 0), ("translate", [], 0), ("merge", ["translate", "summarise"], 0)], {}),
     ("done", [("summarise", 1), ("translate", 1), ("merge", 1)], None)),
    (([("publish", ["draft"], 0), ("draft", [], 0)], {}), ("done", [("draft", 1), ("publish", 1)], None)),
    # no retries: one failure is final
    (([("charge", [], 0)], {"charge": 1}), ("failed", [("charge", 1)], "charge")),
    (([], {}), ("done", [], None)),
    # deep enough to break a recursive sort at Python's default limit
    ((CHAIN, {}), ("done", [(f"s{i}", 1) for i in range(5000)], None)),
]