Tres errores típicos con async¶

7 minutos de lectura · 15 abril 2026

Son los tres errores que llevo dos años viendo en code reviews y cometiendo yo mismo en los días malos. Ninguno es exótico. Los tres se cuelan en producción con regularidad porque el código parece correcto y los tests pasan.

Async no es paralelismo. Async es cooperación. La mayoría de bugs vienen de tratarlo como lo primero.

Voy con Python (asyncio) pero los tres patrones se traducen 1:1 a JS, Rust y cualquier runtime con tareas concurrentes.

1. await en bucle cuando podría ir en gather¶

El más frecuente. Lo veo en cada base de código que toco:

async def descargar_usuarios(ids):
    resultados = []
    for id in ids:
        u = await fetch(id)            # serial
        resultados.append(u)
    return resultados

Si las descargas son independientes —y casi siempre lo son— esto es serial sin razón. Cien usuarios a 200ms son veinte segundos. La versión que aprovecha asyncio:

async def descargar_usuarios(ids):
    return await asyncio.gather(*(fetch(id) for id in ids))

Veinte segundos pasan a doscientos milisegundos. Cuando el bucle es independiente, gather. Cuando depende del anterior, deja el await en el bucle.

Aviso: gather sin límite contra una API ajena es una receta para 429. Usa un Semaphore o asyncio.TaskGroup con control de concurrencia:

sem = asyncio.Semaphore(10)
async def fetch_limitado(id):
    async with sem:
        return await fetch(id)

2. Mezclar código bloqueante con async¶

async def procesar(archivo):
    contenido = open(archivo).read()        # bloquea el loop
    hash = hashlib.sha256(contenido).hexdigest()  # también, si es grande
    return hash

open().read() no es awaitable. Bloquea el event loop entero. Una request lenta puede congelar a todos los demás clientes del servicio.

Soluciones, en orden de preferencia:

1. Usar una librería async nativa: aiofiles, httpx, asyncpg.
2. Para llamadas síncronas inevitables, asyncio.to_thread(...) aísla el bloqueo en un thread del pool:

   contenido = await asyncio.to_thread(_leer_sync, archivo)
   

3. Para CPU pesada (no I/O): ProcessPoolExecutor. Threads no ayudan con el GIL.

La regla: dentro de un async def, cada llamada o es await, o tarda microsegundos. Si tarda milisegundos y no es await, es un bug latente.

3. Exception swallowing en create_task¶

asyncio.create_task(enviar_email(u))   # fire-and-forget

Si enviar_email lanza una excepción y nadie hace await de esa task, la excepción muere silenciosa al recolectarse la task por el garbage collector. A veces sale un warning. A veces no.

El patrón seguro desde Python 3.11 es asyncio.TaskGroup:

async with asyncio.TaskGroup() as tg:
    for u in usuarios:
        tg.create_task(enviar_email(u))
# si alguna falló, salta aquí con ExceptionGroup

Si necesitas fire-and-forget de verdad (job en background que no debe romper el flujo), al menos engánchate al resultado:

task = asyncio.create_task(enviar_email(u))
task.add_done_callback(lambda t: log_si_falla(t.exception()))

Nunca dejes una task huérfana sin nadie observando su excepción.

¿Comentarios o correcciones? info@encodigo.es.