并发竞速返回首个成功结果

功能说明

这段代码实现“对冲请求”模式：把同一操作同时发给多个副本，谁先成功就返回谁的结果，其余任务立即取消。它适合读多写少、延迟敏感且允许重复查询的场景，例如从多个缓存节点或只读副本读取同一份数据。

与逐个重试不同，竞速会并行消耗资源，但能把尾延迟从“最慢副本”降到“最快副本”。所有副本都失败时，函数会汇总错误信息并抛出 RuntimeError；整个竞速还受一个总截止时间约束。

关键代码

first_success 先用 create_task 立即启动所有可等待对象，再用一个 deadline 记录总超时。循环调用 asyncio.wait(..., return_when=FIRST_COMPLETED)，每次只等待已启动任务中最早完成的一批。

对每个完成的任务调用 task.exception()：没有异常就返回结果；有异常则收集起来。finally 中取消所有仍挂起的任务，并用 gather(..., return_exceptions=True) 等待取消完成，避免“任务已取消但仍在运行”的泄漏。

使用注意

任务一旦启动就会产生副作用，取消只能停止协程，无法撤销已经发出的请求或写入。被取消的协程必须正确响应 CancelledError，否则会延迟清理。竞速不适合非幂等写操作，否则多个副本可能同时执行。

总超时是整体预算，不是单个任务的超时；如果某个副本很慢但其他副本都失败，函数会在预算耗尽时抛 TimeoutError。错误汇总只保留异常文本，生产环境应记录结构化上下文，并注意不要泄露敏感信息。

验证方式

示例中 broken 在 0.01 秒失败，fast 在 0.05 秒成功，slow 需要 0.2 秒，因此返回 fast，且慢任务会在 finally 中被取消。第二个场景用事件确认被取消的任务确实收到 CancelledError。两个副本都失败时会打印包含两个错误信息的 RuntimeError。

代码示例

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Iterable
from typing import TypeVar

T = TypeVar("T")


async def first_success(awaitables: Iterable[Awaitable[T]], *, timeout: float) -> T:
    tasks = [asyncio.create_task(awaitable) for awaitable in awaitables]
    if not tasks:
        raise ValueError("at least one awaitable is required")

    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    pending = set(tasks)
    failures: list[BaseException] = []
    try:
        while pending:
            remaining = deadline - loop.time()
            if remaining <= 0:
                raise TimeoutError("no replica succeeded before the deadline")
            done, pending = await asyncio.wait(
                pending,
                timeout=remaining,
                return_when=asyncio.FIRST_COMPLETED,
            )
            if not done:
                raise TimeoutError("no replica succeeded before the deadline")
            for task in done:
                error = task.exception()
                if error is None:
                    return task.result()
                failures.append(error)
        messages = "; ".join(str(error) for error in failures)
        raise RuntimeError(f"all replicas failed: {messages}")
    finally:
        for task in pending:
            task.cancel()
        if pending:
            await asyncio.gather(*pending, return_exceptions=True)


async def fetch(name: str, delay: float, *, fail: bool = False) -> str:
    await asyncio.sleep(delay)
    if fail:
        raise ConnectionError(f"{name} unavailable")
    return name


async def main() -> None:
    result = await first_success(
        [fetch("slow", 0.20), fetch("fast", 0.05), fetch("broken", 0.01, fail=True)],
        timeout=1.0,
    )
    print(f"winner: {result}")
    assert result == "fast"

    cancelled = asyncio.Event()

    async def cancellable() -> str:
        try:
            await asyncio.sleep(30)
        except asyncio.CancelledError:
            cancelled.set()
            raise
        return "never"

    assert await first_success(
        [fetch("quick", 0.01), cancellable()],
        timeout=1.0,
    ) == "quick"
    await asyncio.sleep(0)
    assert cancelled.is_set()

    try:
        await first_success(
            [fetch("a", 0.01, fail=True), fetch("b", 0.02, fail=True)],
            timeout=1.0,
        )
    except RuntimeError as exc:
        print(f"all failed: {exc}")


if __name__ == "__main__":
    asyncio.run(main())