Work queue with a dead-letter queue and dependency cycle detection

Three parts, one state machine. Write it down before coding; every rule in the prompt is one row of it.

Part 1: dependencies (Kahn's algorithm, online). For each task keep waiting[t], the number of its dependencies that have not succeeded. A task with waiting == 0 is READY and joins a FIFO deque. When a task completes, walk its dependents and decrement; any that reach 0 become READY. A dependency that has not been submitted yet simply counts as not succeeded, so tasks can arrive in any order. Each edge is touched once on submit and once on completion: O(V + E) for the whole run, the same as an offline topological sort.

Retries. attempts counts how many times a task was handed out. fail puts the task at the back of the ready line while attempts < max_attempts, and kills it otherwise. Back of the line, not the front: a task that fails fast must not starve everything behind it.

Part 2: cycle detection on submit. A cycle can only appear through the task being added, because the graph was acyclic before. So the check is one question: can the new task reach itself? That is, starting from its dependencies and following dependency edges, do we arrive back at the new task? One DFS, O(V + E) worst case, and usually tiny because it only explores the new task's ancestors. Reject the submit with an error and change nothing, so the caller can fix the graph and resubmit.

The tricky case is a cycle closed through a task submitted earlier with a not-yet-known dependency: a waits on b, b waits on c, then c arrives waiting on a. The DFS from c's dependencies goes a → b → c and finds it, because a and b stored their dependency lists when they were submitted. A check that only looked at known tasks' states would miss this and leave three tasks PENDING forever, which is the silent failure the interviewer is probing for.

Part 3: dead letters.

• A task out of attempts becomes DEAD and enters the dead-letter queue (DLQ) with its last error.
• Everything downstream of it can never run. Leaving those tasks PENDING is a silent hang: the queue looks idle and nobody is told. Mark every transitive dependent BLOCKED and put it in the DLQ too, with the reason naming the dead task at the root (an operator wants "blocked by fetch", not "blocked by parse").
• A task submitted later onto a dead or blocked dependency is blocked at once.
• redrive(task) gives a DEAD task fresh attempts and puts it back in line. Then walk its dependents: a BLOCKED task becomes PENDING again only if none of its other dependencies is still DEAD or BLOCKED. If one is, it stays blocked and its reason is updated to name the other cause. Re-queue a task's dependents every time it is unblocked, so a diamond resolves no matter which branch the walk reaches first.

from __future__ import annotations

from collections import defaultdict, deque

PENDING, READY, RUNNING = "PENDING", "READY", "RUNNING"
SUCCEEDED, DEAD, BLOCKED = "SUCCEEDED", "DEAD", "BLOCKED"


class CycleError(ValueError):
    pass


class WorkQueue:
    def __init__(self, max_attempts: int) -> None:
        if max_attempts < 1:
            raise ValueError("max_attempts must be >= 1")
        self.max_attempts = max_attempts
        self.state: dict = {}
        self.deps: dict = {}                      # task -> its dependencies
        self.dependents = defaultdict(list)       # task -> tasks that wait on it
        self.waiting: dict = {}                   # task -> deps not yet SUCCEEDED
        self.attempts: dict = {}
        self.ready: deque = deque()               # READY tasks, FIFO
        self.dlq: dict = {}                       # task -> reason, in arrival order

    # -- level 1 -------------------------------------------------------------
    def submit(self, task: str, deps=()) -> bool:
        if task in self.state:
            return False
        deps = list(dict.fromkeys(deps))          # drop repeats, keep order
        if task in deps or self._reaches(deps, task):
            raise CycleError(f"{task} would depend on itself")
        self.state[task] = PENDING
        self.deps[task] = deps
        self.attempts[task] = 0
        self.waiting[task] = sum(1 for d in deps if self.state.get(d) != SUCCEEDED)
        for d in deps:
            self.dependents[d].append(task)
        bad = next((d for d in deps if self.state.get(d) in (DEAD, BLOCKED)), None)
        if bad is not None:
            self._block(task, bad)
        elif self.waiting[task] == 0:
            self._make_ready(task)
        return True

    def next_ready(self) -> str | None:
        if not self.ready:
            return None
        task = self.ready.popleft()
        self.state[task] = RUNNING
        self.attempts[task] += 1
        return task

    def complete(self, task: str) -> bool:
        if self.state.get(task) != RUNNING:
            return False
        self.state[task] = SUCCEEDED
        for t in self.dependents[task]:
            self.waiting[t] -= 1
            if self.waiting[t] == 0 and self.state[t] == PENDING:
                self._make_ready(t)
        return True

    def fail(self, task: str, error: str = "") -> str | None:
        if self.state.get(task) != RUNNING:
            return None
        if self.attempts[task] < self.max_attempts:
            self._make_ready(task)                # retry at the back of the line
            return READY
        self.state[task] = DEAD
        self.dlq[task] = f"failed after {self.attempts[task]} attempts: {error}"
        for t in self._downstream(task):
            if self.state[t] == PENDING:  # downstream of a task that never succeeded
                self._block(t, task)
        return DEAD

    def status(self, task: str) -> str | None:
        return self.state.get(task)

    # -- level 3: dead letters -------------------------------------------------
    def dead_letters(self) -> list:
        return list(self.dlq.items())

    def redrive(self, task: str) -> bool:
        """Give a DEAD task a fresh set of attempts and unblock what waited on it."""
        if self.state.get(task) != DEAD:
            return False
        del self.dlq[task]
        self.attempts[task] = 0
        self._make_ready(task)
        todo = deque(self.dependents[task])
        while todo:
            t = todo.popleft()
            if self.state.get(t) != BLOCKED:
                continue
            bad = next((d for d in self.deps[t] if self.state.get(d) in (DEAD, BLOCKED)), None)
            if bad is not None:
                self._block(t, bad)               # still blocked: name the other cause
                continue
            del self.dlq[t]
            self.state[t] = PENDING
            todo.extend(self.dependents[t])
        return True

    # -- helpers -----------------------------------------------------------------
    def _make_ready(self, task: str) -> None:
        self.state[task] = READY
        self.ready.append(task)

    def _block(self, task: str, dep: str) -> None:
        # Name the dead task at the root, even through a chain of blocked ones.
        self.state[task] = BLOCKED
        self.dlq[task] = self.dlq[dep] if self.state[dep] == BLOCKED else f"blocked by {dep}"

    def _downstream(self, task: str) -> list:
        """Every task that depends on `task`, directly or not, nearest first."""
        seen, order, todo = {task}, [], deque([task])
        while todo:
            for t in self.dependents[todo.popleft()]:
                if t not in seen:
                    seen.add(t)
                    order.append(t)
                    todo.append(t)
        return order

    def _reaches(self, starts, target: str) -> bool:
        """Is `target` reachable from `starts` along dependency edges?"""
        seen, stack = set(), list(starts)
        while stack:
            t = stack.pop()
            if t == target:
                return True
            if t in seen:
                continue
            seen.add(t)
            stack.extend(self.deps.get(t, ()))
        return False

V tasks, E dependency edges.

Worked numbers. The cycle DFS on each submit visits only the new task's ancestors. For a wide, shallow DAG (1,000 shards that each wait on one download and feed one reducer) that is one or two tasks per submit, so the whole graph costs about 2,000 steps. For a chain of 10^5 tasks, submit i visits i ancestors: 10^5 × 10^5 / 2 = 5 × 10^9 steps, which is minutes in Python. Say which shape you expect; if graphs can be deep, use the incremental approach in the follow-ups.

• "Make the cycle check cheap for deep graphs." Keep a topological order index per task (a number) and maintain it incrementally (the Pearce–Kelly algorithm): adding edge u → v is only a problem if ord[v] < ord[u], and then you search only the tasks whose order lies between them. Most inserts are O(1). Or batch: accept a whole DAG per submit call and run one Kahn's pass over it, rejecting the batch if anything is left with a nonzero in-degree.
• "Workers crash." Add leases: next_ready returns a lease that expires after T seconds unless heartbeated, and an expired lease counts as a failure (see the leases-and-retries version of this question). Then complete must carry the lease id, or a slow worker can complete a task that was already handed to someone else.
• "Retry with backoff." Failed tasks wait base × 2^attempt (with jitter) before they are READY again: a delay heap in front of the ready deque. Distinguish retryable errors (timeouts, 503) from permanent ones (bad input): a permanent error goes straight to the DLQ without burning attempts.
• "Operate the DLQ." Alert on DLQ growth, not on each entry. Redrive in bulk by reason (redrive where reason startswith "HTTP 503") after the dependency recovers. Keep a poison-pill cap: a task redriven three times goes to a human queue instead.
• "Make it durable and distributed." Store tasks and edges in Postgres; next_ready is SELECT ... WHERE state = 'READY' ORDER BY ready_at LIMIT 1 FOR UPDATE SKIP LOCKED; complete and the dependent decrements happen in one transaction so a crash cannot leave a counter half-updated. At very high volume, shard by DAG id so a DAG's edges stay in one shard.

Mid-level (L4)

• Correct dependency ordering with in-degree counts, FIFO among ready tasks, and retries to the back of the line.
• Detects a direct cycle (and self-dependency) and rejects the submit without changing state.
• Tests for ordering, retry and dead.

Senior (L5)

• Catches the cycle closed through a task submitted earlier with an unknown dependency, and explains why checking only from the new task is enough.
• Blocks every downstream task when a task dies (no silent hang) and records a root-cause reason in the DLQ.
• Redrive that respects diamonds and other dead dependencies; a randomized DAG test.

Staff (L6+)

• States the cost of the cycle check on deep graphs and offers incremental topological ordering or batch submits.
• Brings leases, backoff with retryable versus permanent errors, and DLQ operations (alerting, bulk redrive, poison pills).
• Sketches the durable version: transactional state changes, SKIP LOCKED dispatch, sharding by DAG.