Timeouts and Failures

Networks fail in two ways: something goes wrong and you get an error, or nothing happens at all, forever. The first is easy to notice. The second is worse, because a program waiting for an answer that never comes looks exactly like a program that’s working. So every wait on a network needs a limit.

Time limits

asyncio.timeout(seconds) puts a deadline on everything inside its block. When time runs out, the task is cancelled and the block raises TimeoutError. asyncio.wait_for does the same for a single awaitable:

import asyncio


async def fetch_log(server, delay):
    print(f"{server}: connecting")
    try:
        await asyncio.sleep(delay)
        return f"{server} log"
    finally:
        print(f"{server}: connection closed")     # runs on success, failure or cancellation


async def main():
    try:
        async with asyncio.timeout(0.2):
            print(await fetch_log("slow", 1.0))
    except TimeoutError:
        print("slow: gave up after 0.2s")

    print(await asyncio.wait_for(fetch_log("fast", 0.1), timeout=0.2))


asyncio.run(main())

Look at the order of the output for slow. The connection was closed before “gave up” was printed. Here’s what happened when the timeout expired:

1. The event loop cancelled the task, by raising asyncio.CancelledError inside it, at the await where it was paused.
2. That exception ran the finally block on its way out, like any exception would, so the connection was closed.
3. The timeout block turned the cancellation into a TimeoutError for the caller.

This is why the finally rule from Day 4 matters even more in async code. A coroutine can be stopped at any await, and cleanup that isn’t in a finally (or a with) just doesn’t happen.

When one of several fails

With three sources, one of them failing is normal. What should happen to the other two? asyncio offers three different answers:

import asyncio


async def fetch_log(server, delay, fails=False):
    await asyncio.sleep(delay)
    if fails:
        raise ConnectionError(f"{server}: connection refused")
    return f"{server} log"


async def main():
    print("gather, return_exceptions=True:")
    results = await asyncio.gather(fetch_log("auth", 0.2), fetch_log("db", 0.1, fails=True),
                                   return_exceptions=True)
    for result in results:
        print("  ", repr(result))

    print("gather, default:")
    try:
        await asyncio.gather(fetch_log("auth", 0.2), fetch_log("db", 0.1, fails=True))
    except ConnectionError as error:
        print("  ", repr(error), "- and auth's result is gone")

    print("TaskGroup:")
    try:
        async with asyncio.TaskGroup() as group:
            auth = group.create_task(fetch_log("auth", 0.2))
            group.create_task(fetch_log("db", 0.1, fails=True))
    except* ConnectionError as group_error:
        print("  ", group_error.exceptions)
        print("   auth was cancelled:", auth.cancelled())


asyncio.run(main())

ExceptionGroup holds several exceptions at once, because in concurrent code more than one thing can fail at the same time. except* catches the matching exceptions inside a group. except* ConnectionError handles every ConnectionError, and anything else in the group carries on up.

TaskGroup’s all-or-nothing rule is the right default when the tasks depend on each other: if one part of a job fails, the rest is wasted work. Whether it’s right for your tasks is a decision, as the next bug hunt shows.

Bug hunt: one server down, no report at all

import asyncio

DELAYS = {"auth": 0.3, "payments": 0.2, "search": 0.1}
DOWN = {"payments"}                                   # today, payments is down


async def fetch_log(server):
    await asyncio.sleep(DELAYS[server])
    if server in DOWN:
        raise ConnectionError(f"{server}: connection refused")
    return [f"2026-09-01 09:00:00 ERROR {server}: x=1"] * 2


async def collect(servers):
    async with asyncio.TaskGroup() as group:
        tasks = {server: group.create_task(fetch_log(server)) for server in servers}
    return {server: task.result() for server, task in tasks.items()}


logs = asyncio.run(collect(DELAYS))
for server, lines in logs.items():
    print(f"{server}: {len(lines)} lines")

Reproduce. One server out of three is down, and the program crashes with an ExceptionGroup. No report at all, not even for the two servers that worked.

Inspect. What happened to each task? Replace the last three lines with a version that catches the group and looks at every task:

async def inspect():
    tasks = {}
    try:
        async with asyncio.TaskGroup() as group:
            tasks = {server: group.create_task(fetch_log(server)) for server in DELAYS}
    except* ConnectionError:
        pass
    for server, task in tasks.items():
        state = "cancelled" if task.cancelled() else ("failed" if task.exception() else "done")
        print(server, state)

asyncio.run(inspect())

search was done, payments failed, and auth was cancelled while it was still downloading.

Your turn, before you open this. Change collect so the two working servers are reported and payments shows as failed, and write a test that pins it. Then open the reveal and read it either way: it explains which failures should cancel the other fetches, and which shouldn’t.

def test_one_server_down_does_not_lose_the_others(self):
    results = asyncio.run(collect(DELAYS))
    self.assertEqual([r.status for r in results], ["ok", "failed", "ok"])

from dataclasses import dataclass


@dataclass(frozen=True)
class Result:
    server: str
    status: str                       # "ok" or "failed"
    lines: tuple[str, ...] = ()
    error: str = ""


async def fetch_safely(server):
    try:
        return Result(server, "ok", tuple(await fetch_log(server)))
    except ConnectionError as error:
        return Result(server, "failed", error=str(error))


async def collect(servers):
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(fetch_safely(server)) for server in servers]
    return [task.result() for task in tasks]


for result in asyncio.run(collect(DELAYS)):
    print(f"{result.server}: {result.status}, {len(result.lines)} lines {result.error}")

Your turn

import asyncio
import sys
import unittest
from dataclasses import dataclass

closed = []                                 # servers whose connection was closed


@dataclass(frozen=True)
class Result:
    server: str
    status: str                             # "ok", "failed" or "timeout"
    lines: tuple[str, ...] = ()
    error: str = ""


async def fetch_log(server):
    try:
        delay = {"fast": 0.05, "slow": 1.0, "down": 0.05, "buggy": 0.05}[server]
        await asyncio.sleep(delay)
        if server == "down":
            raise ConnectionError("connection refused")
        if server == "buggy":
            raise KeyError("a bug, not a network problem")
        return [f"{server}: line {i}" for i in range(3)]
    finally:
        closed.append(server)


async def fetch_safely(server, timeout):
    raise NotImplementedError("your code here")


class TestFetchSafely(unittest.TestCase):
    def test_ok(self):
        result = asyncio.run(fetch_safely("fast", timeout=0.5))
        self.assertEqual((result.status, len(result.lines)), ("ok", 3))

    def test_failed(self):
        result = asyncio.run(fetch_safely("down", timeout=0.5))
        self.assertEqual((result.status, result.error), ("failed", "connection refused"))

    def test_timeout_closes_the_connection(self):
        closed.clear()
        result = asyncio.run(fetch_safely("slow", timeout=0.1))
        self.assertEqual((result.status, closed), ("timeout", ["slow"]))

    def test_bugs_are_not_hidden(self):
        with self.assertRaises(KeyError):
            asyncio.run(fetch_safely("buggy", timeout=0.5))


suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestFetchSafely)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

async def fetch_safely(server, timeout):
    try:
        async with asyncio.timeout(timeout):
            lines = await fetch_log(server)
    except TimeoutError:
        return Result(server, "timeout", error=f"no answer after {timeout}s")
    except ConnectionError as error:
        return Result(server, "failed", error=str(error))
    return Result(server, "ok", tuple(lines))

What you can now do

• Put a deadline on any await with asyncio.timeout or asyncio.wait_for.
• Explain cancellation, and make cleanup happen with finally.
• Choose between gather, gather(return_exceptions=True) and TaskGroup according to what one failure should mean.
• Handle expected failures per task, as data, while letting bugs escape.

Next: the Day 6 checkpoint, four sources at once, using everything from today.