Async worker pool with retries · Python practice

Async worker pool with retries

Hard~30 min

Process items with exactly workers concurrent asyncio worker tasks reading from an asyncio.Queue. For each item call the provided await process(item):

On TransientError retry the item (up to max_attempts attempts in total).
On any other exception, or when attempts run out, record the error type name.

Return {"results": [...], "errors": {...}} where results[i] is the result for items[i] (or None if it failed) and errors maps the string index to the exception type name. Shut the workers down cleanly (no pending tasks). The harness checks the concurrency you achieved.

Function

Implement run_pipeline. Input: items: list[str], workers: int, max_attempts: int. Returns: dict.

Examples

>>> run_pipeline(['a', 'flaky:3', 'bad:1', 'flaky:5'], 2, 3)

{'clean': True, 'errors': {'2': 'ValueError', '3': 'TransientError'}, 'peak_ok': True, 'results': ['A', 'FLAKY:3', None, None]}

flaky:3 fails twice and succeeds on attempt 3; flaky:5 runs out of attempts.

Provided environment

This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.

import asyncio

class TransientError(Exception):
    pass

_state = {"active": 0, "peak": 0, "attempts": {}}

async def process(item):
    _state["active"] += 1
    _state["peak"] = max(_state["peak"], _state["active"])
    try:
        n = _state["attempts"][item] = _state["attempts"].get(item, 0) + 1
        await asyncio.sleep(0.002)
        kind, _, rest = item.partition(":")
        if kind == "flaky" and n < int(rest.split(":")[0]):
            raise TransientError(item)
        if kind == "bad":
            raise ValueError(item)
        return item.upper()
    finally:
        _state["active"] -= 1

async def __harness__(fn, items, workers, max_attempts):
    _state.update(active=0, peak=0, attempts={})
    out = await fn(items, workers, max_attempts)
    leftover = [t for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()]
    return {**out, "peak_ok": _state["peak"] == min(workers, len(set(items))) or not items, "clean": not leftover}

Details

Level: Concurrency, resilience patterns, performance constraints and production-style automation.
Roles: AI Engineer, Backend Engineer, Data Engineer
Topics: asyncio, Retries and backoff, Threads and concurrency
Tests: 1 public (run with Run), 3 hidden (graded on Submit)
Grading: Exact match of the return value