Workspace/Mini projects
Loading progress
All mini projects
MODULE 14 · 5 HOUR BUILD

Recoverable release review coordinator

Build a small coordinator that executes independent review tasks, persists task outputs, and recovers without duplicating a final external action. The seed is an in-memory recovery simulation.

Build evidence Record your actual checks, results, and limitations.

Build it in stages

  1. Define schemas for review inputs, evidence rows, and a merged report.
  2. Replace simulated review functions with fixture-backed independent checks.
  3. Add bounded concurrent execution and explicit partial-failure policy.
  4. Persist checkpoints and an operation ledger with versioned schemas.
  5. Inject crashes before and after the simulated sink and document recovery traces.

Your acceptance criteria

Use these as your project review. Record commands, outputs, and failure cases in your repository.

  • A repeated run produces one logical publication for the same run and plan version.
  • A changed payload under an existing operation key is rejected.
  • A failed worker is distinguishable from a worker with no findings.
  • A report records every input version and unresolved conflict.

A working starting point

The seed runs as supplied. Extend it to satisfy the full brief. It is a teaching starting point, not a finished portfolio submission.

main.py
python
import json
from graphlib import TopologicalSorter

GRAPH = {'api': (), 'tests': (), 'report': ('api', 'tests')}
SINK = {}

def publish(key, payload):
    encoded = json.dumps(payload, sort_keys=True)
    if key in SINK and SINK[key] != encoded:
        raise ValueError('changed operation payload')
    SINK.setdefault(key, encoded)

def execute(state, crash_after_publish=False):
    for task in TopologicalSorter(GRAPH).static_order():
        if task in state['done']:
            continue
        if task == 'api':
            result = {'check': 'api', 'findings': ['deprecated alias']}
        elif task == 'tests':
            result = {'check': 'tests', 'findings': []}
        else:
            result = {'reviews': [state['outputs'][key]
                                  for key in ('api', 'tests')]}
            publish('release-17:report:v1', result)
            if crash_after_publish:
                return state
        state['outputs'][task] = result
        state['done'].append(task)
    return state

def main():
    state = {'schema': 1, 'done': [], 'outputs': {}}
    execute(state, crash_after_publish=True)
    state = json.loads(json.dumps(state))
    execute(state)
    execute(state)
    print('completed:', ','.join(state['done']))
    print('logical publications:', len(SINK))
    print('findings:', sum(len(row['findings'])
                           for row in state['outputs']['report']['reviews']))

if __name__ == '__main__':
    main()

Push it further

Implement a resource-aware scheduler, then compare one-worker and multiworker execution using the same cases, costs, and correctness rubric.