Design and incrementally build a **streaming job scheduler**: a service (and supporting class) that continuously ingests tasks and dispatches them to workers, respecting dependencies and deadlines. This was a coding round that progressively grew into a mini system design. You will first implement a single-process class, then add dependency handling, then discuss how it becomes a distributed, fault-tolerant service. Each task is defined as: - `id`: unique task identifier - `deadline`: an execution deadline (orderable, e.g. an epoch timestamp) - `prerequisites`: a list of task `id`s that must complete before this task can run (empty for the first two parts) ### Constraints & Assumptions - **Ordering rule:** among *runnable* tasks (all prerequisites completed), the scheduler dispatches the one with the **earliest deadline** first; ties broken deterministically (e.g. by `id`). - **Scale (Part 3):** assume large task volumes and many concurrent workers, with ingest arriving in bursts. - **At-least-once execution** is acceptable for individual tasks (workers must be idempotent), but the **same task must never be leased to two workers simultaneously**. - For Parts 1–2 a **single-process, single-threaded** implementation is fine. Concurrency, persistence, and distribution enter only in Part 3. - The interview allowed Python or Go; pseudocode is acceptable here as long as the data structures and complexity are explicit. ### Clarifying Questions to Ask - Can a task reference a prerequisite that **hasn't been submitted yet** (forward references), or must every prerequisite already exist at submission time? - Are tasks grouped into independent **workflows/DAGs**, or is there one global dependency graph? (This decides the blast radius of cycle checks and partitioning.) - Is **at-least-once** execution acceptable (workers idempotent), or do we need exactly-once semantics? - What should happen to a task's **dependents when a prerequisite permanently fails** — cancel the subtree, or skip and continue? - Is the deadline a **hard SLA** (drop/alert if missed) or merely a priority hint? - What are the expected ingest rate, total live-task count, and max fan-in/fan-out per task? --- ### Part 1 — In-memory scheduler with deadline ordering Implement a `Scheduler` class with two operations (no prerequisites yet): - `add_tasks(tasks)`: accept one task or a batch of tasks. - `consume_task()`: remove and return the task with the **earliest deadline**. If no task is available, return a sentinel such as `"no task"` (or `None`). State the time complexity of each operation. ```hint What does "earliest deadline first" demand? Both operations need to repeatedly find the *minimum* `deadline` among everything currently held. Which classic data structure makes "insert" and "extract-min" both cheap, and what does that imply for the per-operation complexity you're asked to state? ``` ```hint Determinism and later lookups Two tasks can share a `deadline`; think about what to add to the ordering key so the result is total and reproducible. Separately, if a later part needs to find or remove a specific task by `id`, ask yourself whether your chosen structure can address an element by key — and what auxiliary structure would cover that gap. ``` #### What This Part Should Cover - The right ordering primitive for repeated extract-min, with the **stated complexity** of each operation and the memory footprint. - A **deterministic, total tie-break** baked into the ordering key (not left to undefined heap behavior). - Awareness that the chosen structure can't address elements by key, and the auxiliary index that anticipates later needs (lookup, cancel, dedupe). - A clear, sensible contract for the empty case. --- ### Part 2 — Add prerequisites (dependency-aware dispatch) Each task now carries `prerequisites`. A task becomes **runnable** only after *all* of its prerequisites have completed. `consume_task()` must return the earliest-deadline task **among runnable tasks only** — never a task whose prerequisites are still pending. You also need a `complete_task(id)` operation so the scheduler learns a task finished and can unblock its dependents. Explain how you keep this efficient as tasks complete, and how the structure relates to a topological order. ```hint Avoid recomputing readiness Recomputing "are all prerequisites done?" by scanning the dependency graph on every `consume` is wasteful. What single per-task counter could you maintain so that "runnable" becomes an $O(1)$ check, and when would that counter change? ``` ```hint Reacting to a completion cheaply When a task finishes, only a small set of other tasks can possibly become newly runnable. Which set, and what would you need to store at insert time so that on completion you can reach exactly those tasks without a graph walk? Think about which direction your edges should point — the direction you read edges for dispatch is not necessarily the direction you need when propagating a completion. ``` ```hint Contracts and edge cases Consider a task whose prerequisites are already satisfied the moment it arrives — where should it land? And decide your contract for a prerequisite `id` you have never seen: is it a forward reference you'll wait for, or an error you reject? State the rule; it changes both the code and the validation. ``` #### Clarifying Questions for this Part - When a task's prerequisites are **already completed** at insert time, should it become runnable immediately, or wait for an explicit trigger? - Can the same prerequisite edge appear more than once, and should duplicate edges be collapsed or rejected? #### What This Part Should Cover - An **incremental readiness scheme** — a per-task counter of incomplete prerequisites that makes "runnable" an $O(1)$ check rather than a graph scan. - The correct **edge direction** (reverse adjacency) so a completion reaches exactly its dependents without walking the graph, plus the per-completion and end-to-end complexity. - An explicit connection to **Kahn's topological sort**: the runnable frontier is the zero-indegree set, ordered here by deadline. - A **stated contract** for forward references (wait vs. reject) that the code and validation agree on, plus the immediately-runnable edge case. --- ### Part 3 — Validation, DAG verification, concurrency, and streaming (mini system design) Now turn the class into a service. Beyond Parts 1–2, the system must: 1. Accept tasks in **batches or as a continuous stream**. 2. **Validate** task definitions: malformed input, duplicate `id`s, self-dependencies, and prerequisite references that don't (and never will) resolve. 3. **Verify the dependency graph is a DAG** — reject submissions that would introduce a cycle. 4. Dispatch only runnable tasks (Part 2 semantics) in earliest-deadline order. 5. Support **many workers consuming concurrently** without ever assigning the same task twice. 6. Handle **worker crashes, retries, and recovery** (a dispatched task whose worker dies must become runnable again). 7. **Scale** to the volumes above and provide **useful monitoring**. Discuss API design, storage model, scheduling logic, DAG-validation strategy (batch vs. streaming), concurrency control, fault tolerance, and the scaling approach. ```hint DAG check: batch vs. streaming For a full batch/workflow submitted together, cycle detection is **Kahn's algorithm** or DFS coloring in $O(V+E)$. For *streaming* edge-by-edge inserts, checking "does adding $u \to v$ create a cycle?" reduces to "is $u$ already reachable from $v$?" — a bounded DFS/BFS in the affected workflow. Note the cost difference and consider scoping cycle checks per `workflow_id`. ``` ```hint Assigning tasks without double-dispatch In a single process, "pop" is fine — only one thread touches the heap. In a distributed setting multiple workers race to consume the same runnable row. What property must the "take a task" operation have so that two workers can never both succeed on the same task? What class of database or cache primitive provides that property? ``` ```hint Worker crashes and replay safety Suppose a worker takes a task and then dies before completing it. How does the service know the task needs to be re-dispatched? And if the same complete/fail message arrives twice (e.g. a delayed retry from a worker whose lease already moved on), how do you ensure that dependents are unblocked exactly once rather than twice? ``` ```hint Scaling the dispatch path If all tasks lived in one global runnable queue, every scheduler instance would contend on the same structure. What property of the data (available from the task definition itself) could you use to split the runnable set into independent sub-queues so that most operations require no cross-queue coordination? ``` #### Clarifying Questions for this Part - Is there an existing message bus / queue (Kafka, SQS) and a durable store (SQL, Redis) we should build on, or are we choosing the stack? - For streaming submissions, may we require a **complete DAG per workflow** at submit time, or must we accept genuinely edge-at-a-time inserts into an open workflow? - What recovery-time and durability targets apply (how fast must a crashed worker's task be re-dispatched, how much in-flight state may we lose)? #### What This Part Should Cover - A **clean task state machine** (`WAITING → RUNNABLE → LEASED → COMPLETED / FAILED`) and where each transition happens, mapped back to the Part-1/2 primitives. - A **durable storage model** (task metadata, edge/adjacency table, an externalized runnable index, an ingest log) and a focused **API surface** (batch add, lease, complete, fail, heartbeat). - **Atomic leasing** that provably prevents double-assignment, naming the concrete primitive (`FOR UPDATE SKIP LOCKED` / Lua `ZPOPMIN`) rather than hand-waving. - An explicit **batch-vs-streaming DAG-validation tradeoff** with the cost of each and a pragmatic restriction, not just "run Kahn's". - **Fault tolerance**: lease-expiry recovery via a sweeper, retries with backoff + attempt cap, idempotent submit/complete with fence tokens, and rebuilding state from the durable store on restart. - A **scaling and observability** story: partition-by-workflow rationale, per-shard runnable queues, separation of ingest/schedule/completion paths, and the metrics that expose backlog, starvation, and correctness bugs. --- ### What a Strong Answer Covers These dimensions span all three parts — a top answer carries one coherent design from the in-memory class through to the distributed service. - **One throughline, not three disjoint answers:** the min-heap over the runnable frontier, the indegree / reverse-adjacency unblocking, and the leasing / completion transitions are each shown as the in-memory ancestor of their distributed counterpart. - **Stated complexity and correctness reasoning** at every layer — per-operation costs in Parts 1–2, and a provable no-double-dispatch argument in Part 3. - **Tradeoff awareness:** acknowledging where the general solution is expensive (streaming DAG validation, single-giant-workflow partitioning) and proposing pragmatic restrictions instead of pretending it's free. - **Contract consistency:** the forward-reference / failure-propagation / at-least-once decisions made early are honored uniformly in the code, the validation, and the service semantics. ### Follow-up Questions - A worker takes a lease, then hangs for hours without crashing (a "zombie"). How do you distinguish it from a legitimately slow task, and how do leases + heartbeats handle it without prematurely re-dispatching? - Deadlines keep getting pushed; a low-priority task with a far-future deadline never runs. How would you prevent **starvation** while still honoring earliest-deadline-first? - How do you support **task cancellation** or **deadline updates** for a task already sitting in the runnable heap (which can't address entries by key)? - If a single workflow contains millions of tasks, how does partitioning by `workflow_id` break down, and what would you do instead?