← Python: reusable libraries and packaging Progress & review ↗

Typorithm · medium

Structured cancellation and observable cleanup

pythonlibrariespackagingindependent-practice

Saved on this device

Full lesson · 4 learning steps Open any exercise for focused practice

Step 1 of 4 · intro card

Structured cancellation and observable cleanup

Concept and contract

Coordinate tasks with events, cancel waiting siblings and wait for their finally blocks.

Structured concurrency gives related tasks an owned scope. When one task fails, the scope can cancel its siblings and wait for their completion before leaving. That waiting matters: requesting cancellation alone is not evidence that resource cleanup has actually finished.

The fixture uses asyncio events rather than timing sleeps. Workers announce that they started, then wait indefinitely until a deliberate sibling failure causes cancellation. Each worker records cleanup in finally. The returned report is built only after TaskGroup has exited, so it can verify that every started worker reached its cleanup boundary.

Mechanism and worked example

A shared counter and all_started event coordinate the failure task.

A shared counter and all_started event coordinate the failure task. With two names, both workers reach their waiting point before the failure raises ValueError. The task group cancels those waiting workers, their finally blocks append names and the matching exception-group handler records failure.

Cleanup order may depend on scheduling, so the report sorts names instead of asserting a fragile interleaving. The empty-name case sets the event immediately and still observes the deliberate failure. The synchronous inspect_example entry uses asyncio.run only as a local lab boundary; an async library should expose an awaitable API to an already running event loop.

Complete worked example

python

import asyncio

def inspect_example(names):
    async def run():
        started = 0
        all_started = asyncio.Event()
        cleanup = []
        if not names:
            all_started.set()
        async def worker(name):
            nonlocal started
            started += 1
            if started == len(names):
                all_started.set()
            try:
                await asyncio.Event().wait()
            finally:
                cleanup.append(name)
        async def fail():
            await all_started.wait()
            raise ValueError('controlled sibling failure')
        failed = False
        try:
            async with asyncio.TaskGroup() as group:
                for name in names:
                    group.create_task(worker(name))
                group.create_task(fail())
        except* ValueError:
            failed = True
        return {'closed': sorted(cleanup), 'failed': failed}
    return asyncio.run(run())

Reasoning: A shared counter and all_started event coordinate the failure task. With two names, both workers reach their waiting point before the failure raises ValueError. The task group cancels those waiting workers, their finally blocks append names and the matching exception-group handler records failure.

Download the reusable Python and packaging computer lab ↗

Focused decision: use the complete example for context

python

try:
    await asyncio.Event().wait()
finally:
    cleanup.append(name)

Decision: Record the started worker count and preserve cleanup after a sibling failure cancels every waiting worker.

Download the reusable Python and packaging computer lab ↗

Counterexamples and common mistakes

The controlled bug removes the cleanup append but leaves the task group and failure intact.

The controlled bug removes the cleanup append but leaves the task group and failure intact. The report still says the batch failed, yet no worker cleanup is recorded. That missing evidence is rejected even though the error handler itself looks reasonable.

Swallowing CancelledError can also interfere with structured-concurrency expectations and delay a scope's exit. Finally should perform bounded release work and allow cancellation to propagate unless a carefully documented design genuinely requires otherwise. Avoid proving cancellation with arbitrary short sleeps: scheduling speed is not the contract this deterministic fixture wants to test.

Controlled bug: the executable contract rejects this version

python

import asyncio

def inspect_example(names):
    async def run():
        started = 0
        all_started = asyncio.Event()
        cleanup = []
        if not names:
            all_started.set()
        async def worker(name):
            nonlocal started
            started += 1
            if started == len(names):
                all_started.set()
            try:
                await asyncio.Event().wait()
            finally:
                pass
        async def fail():
            await all_started.wait()
            raise ValueError('controlled sibling failure')
        failed = False
        try:
            async with asyncio.TaskGroup() as group:
                for name in names:
                    group.create_task(worker(name))
                group.create_task(fail())
        except* ValueError:
            failed = True
        return {'closed': sorted(cleanup), 'failed': failed}
    return asyncio.run(run())

Reasoning: The controlled bug removes the cleanup append but leaves the task group and failure intact. The report still says the batch failed, yet no worker cleanup is recorded. That missing evidence is rejected even though the error handler itself looks reasonable.

Download the reusable Python and packaging computer lab ↗

Independent practice

Add the number of started workers to the final report while preserving cleanup and failure observations.

Add the number of started workers to the final report while preserving cleanup and failure observations. The first hint is to reuse the existing counter, which increments before each worker reaches its wait. Do not count only completed successful work: these workers are deliberately cancelled.

The second hint is to compare started with the length of the cleanup list after scope exit. They should agree for the owned fixtures, including the empty case. Keep the event barrier so the failure cannot race ahead of workers and make a partial start look like complete cancellation coverage.

Step 1: Record the started worker count and preserve cleanup after a sibling failure cancels every waiting worker. Step 2: Run the accepted baseline, edit the exercise, then check its independent solution cases; preserve caller inputs and the documented exception behavior.

Practice starter: solve before reading the solution

python

import asyncio

def inspect_example(names):
    async def run():
        started = 0
        all_started = asyncio.Event()
        cleanup = []
        if not names:
            all_started.set()
        async def worker(name):
            nonlocal started
            started += 1
            if started == len(names):
                all_started.set()
            try:
                await asyncio.Event().wait()
            finally:
                cleanup.append(name)
        async def fail():
            await all_started.wait()
            raise ValueError('controlled sibling failure')
        failed = False
        try:
            async with asyncio.TaskGroup() as group:
                for name in names:
                    group.create_task(worker(name))
                group.create_task(fail())
        except* ValueError:
            failed = True
        return {'closed': sorted(cleanup), 'failed': failed}
    return asyncio.run(run())

# Task: Record the started worker count and preserve cleanup after a sibling failure cancels every waiting worker.
# Practice contract: Record the started worker count and preserve cleanup after a sibling failure cancels every waiting worker.

Reasoning: Record the started worker count and preserve cleanup after a sibling failure cancels every waiting worker.

Download the reusable Python and packaging computer lab ↗

Hint 1: Add the number of started workers to the final report while preserving cleanup and failure observations. The first hint is to reuse the existing counter, which increments before each worker reaches its wait. Do not count only completed successful work: these workers are deliberately cancelled.
Hint 2: The second hint is to compare started with the length of the cleanup list after scope exit. They should agree for the owned fixtures, including the empty case. Keep the event barrier so the failure cannot race ahead of workers and make a partial start look like complete cancellation coverage.

Solution, tests and trade-offs

The solution reports started after TaskGroup has waited for sibling cancellation and cleanup.

The solution reports started after TaskGroup has waited for sibling cancellation and cleanup. Acceptance checks two workers, one worker and no workers, preserving the deterministic sorted cleanup list. This verifies scope completion rather than merely dispatching cancellation requests.

Real cleanup can fail, block or need its own timeout and resource policy. TaskGroup cannot make an unbounded finally block finish magically. Separate the ownership guarantee from operational assumptions, and design production tasks so cancellation can be observed at await points while owned resources have explicit, bounded release paths.

Practice solution and acceptance checks

python

import asyncio

def inspect_example(names):
    async def run():
        started = 0
        all_started = asyncio.Event()
        cleanup = []
        if not names:
            all_started.set()
        async def worker(name):
            nonlocal started
            started += 1
            if started == len(names):
                all_started.set()
            try:
                await asyncio.Event().wait()
            finally:
                cleanup.append(name)
        async def fail():
            await all_started.wait()
            raise ValueError('controlled sibling failure')
        failed = False
        try:
            async with asyncio.TaskGroup() as group:
                for name in names:
                    group.create_task(worker(name))
                group.create_task(fail())
        except* ValueError:
            failed = True
        return {'closed': sorted(cleanup), 'failed': failed, 'started': started}
    return asyncio.run(run())

Reasoning: The solution reports started after TaskGroup has waited for sibling cancellation and cleanup. Acceptance checks two workers, one worker and no workers, preserving the deterministic sorted cleanup list. This verifies scope completion rather than merely dispatching cancellation requests.

Download the reusable Python and packaging computer lab ↗

Interview reasoning and next step

Explain why the report is assembled after leaving TaskGroup and why the failure task waits on an event.

Explain why the report is assembled after leaving TaskGroup and why the failure task waits on an event. Then distinguish requesting cancellation, receiving CancelledError and finishing a finally block. These are related stages, not interchangeable completion claims.

For a stretch, add an owned asynchronous resource with a recorded exit and verify it closes for every cancelled worker. Use explicit synchronization to establish the test state. Keep network services and real wall-clock deadlines out of the fixture so a scheduler delay cannot masquerade as a cancellation-policy defect.

Optional stretch — after the core task works: For a stretch, add an owned asynchronous resource with a recorded exit and verify it closes for every cancelled worker. Use explicit synchronization to establish the test state. Keep network services and real wall-clock deadlines out of the fixture so a scheduler delay cannot masquerade as a cancellation-policy defect.

Computer practice: https://applydjinn.com/typorithm/python-engineering-lab-20261107.zip. Use an isolated Python 3.13 environment, install requirements.txt, then run verify.py. Edit exercises/pye-cancellation.py and run --practice pye-cancellation. Fixture files are created in fresh temporary directories for every case. Packaging checks build real local artifacts, install the exact wheel without an index and run outside the source tree. No package is published to PyPI. The app includes static examples, prepared results and questions; it does not run Python or provide an editor.

Coordinate tasks with events, cancel waiting siblings and wait for their finally blocks.