MediumSchedulingPython 3

Workflow State Machine

Track dependent jobs through legal workflow states with retry and cancellation behavior.

40m3 sample tests8 hidden tests

Implement WorkflowRunner, an in-memory job tracker for dependent work. Jobs move through a small set of states:

text
1pending -> running -> succeeded 2pending -> running -> failed 3running -> retrying -> pending 4pending/running/retrying/failed -> canceled

Requirements

  • add_job(job_id, deps=None, max_attempts=1) registers a job.
  • add_job raises ValueError on duplicate ids, max_attempts < 1, or dependencies that are not already registered.
  • ready() returns pending jobs whose dependencies have succeeded, in insertion order.
  • transition(job_id, state, reason=None) validates state transitions and returns the final state.
  • Starting a job increments its attempt count.
  • transition to running is allowed only when the job is currently pending and appears in ready(); otherwise raise ValueError (message must include "dependencies" when the job is blocked on unsatisfied dependencies).
  • Other illegal edges raise ValueError whose message includes "invalid transition".
  • If a running job fails and still has attempts left, move it to retrying; otherwise move it to failed. Requesting "failed" may therefore return "retrying" when attempts remain.
  • retrying jobs are not ready. The caller must explicitly transition(..., "pending") to re-queue them.
  • status(job_id) returns {"state": str, "attempts": int, "history": list[dict]} where each history entry is {"state", "reason", "attempt"} in transition order. attempt is the job's attempt count at the time of that transition.
  • Raise ValueError for invalid transitions or unknown jobs.

Example

python
1runner = WorkflowRunner() 2runner.add_job("extract") 3runner.add_job("train", deps=["extract"], max_attempts=2) 4 5assert runner.ready() == ["extract"] 6runner.transition("extract", "running") 7runner.transition("extract", "succeeded") 8assert runner.ready() == ["train"] 9 10# Multiple ready jobs stay in insertion order (not sorted by id). 11runner2 = WorkflowRunner() 12runner2.add_job("b") 13runner2.add_job("a") 14runner2.add_job("c", deps=["a"]) 15assert runner2.ready() == ["b", "a"]

Constraints

  • Keep it single-process and in memory.
  • Use deterministic ordering.
  • Raise ValueError for invalid transitions or unknown jobs.

Editor