Cómo diseñar el cierre de una ejecución de streaming: lo que permanece después de que se pierde la conexión

La API de transmisión tiene un camino feliz que parece sencillo: iniciar las operaciones externas, enviar fragmentos de resultados y, al final, cambiar el estado de ejecución a succeeded.

Sin embargo, la situación cambia si el usuario cierra la pestaña en medio de la respuesta. La conexión HTTP puede haber terminado, pero la llamada al modelo continúa, o el generador se cierra, pero el registro de ejecución queda como running. Es posible que la pantalla que recibió el evento de finalización y el estado final en la base de datos difieran entre sí.

Para resolver este problema, debemos definir por separado los tiempos de vida de la conexión, la ejecución y el registro persistente, y conectar las responsabilidades de terminación entre ellos. Este artículo parte de una reproducción mínima con generadores asíncronos de Python y extiende el análisis al manejo de la cancelación, los límites de transacción y la propiedad de ejecución en múltiples workers.

Los ejemplos se verificaron en Python 3.12.13. Las explicaciones sobre AnyIO y SQLAlchemy se basan en la documentación de las versiones 4.x y 2.0, respectivamente. Las estructuras de leases de trabajo y almacenamiento de eventos que aparecen posteriormente son ejemplos de diseño, distintos de los experimentos de reproducción de la parte inicial.

1. Primero, separe los cuatro tiempos de vida

No siempre es necesario alinear los cuatro tiempos de vida. En su lugar, se deben definir reglas de vinculación según el comportamiento del producto.

• Ejecución dependiente de la solicitud: Utilice la ruptura de la conexión como señal de cancelación. La responsabilidad de la limpieza recae en el ejecutor que atiende la solicitud.
• Trabajo que debe continuar: Trate la ruptura de la conexión como una desuscripción. La responsabilidad de la ejecución la mantiene un ejecutor independiente de la solicitud.

Si la decisión es ambigua aquí, se generan códigos contradictorios en el mismo programa. Una capa cancela la tarea al cerrar la conexión, mientras que otra espera la notificación de finalización. Aunque ambos son códigos naturales por separado, el contrato del sistema combinado no se sostiene.

2. Salir de async for no cierra inmediatamente el generador aguas arriba

Las transmisiones anidadas suelen estructurarse de la siguiente manera:

외부 응답 생성기 → 실행 이벤트 생성기 → HTTP 응답 생성기

Si cada capa simplemente lee con async for y solo aplica yield, el cierre del generador externo podría no conectar con la terminación explícita del generador interno. Es particularmente importante el caso en que la capa exterior se detiene justo después de pasar un valor.

El siguiente ejemplo verifica el punto de limpieza de los recursos aguas arriba.

import asyncio
from contextlib import aclosing


async def upstream(closed):
    try:
        yield "first chunk"
        await asyncio.Event().wait()
    finally:
        closed.append("upstream closed")


async def relay(source):
    async with aclosing(source) as chunks:
        async for chunk in chunks:
            yield chunk


async def check():
    closed = []
    response = relay(upstream(closed))

    assert await anext(response) == "first chunk"
    await response.aclose()

    # 이벤트 루프 종료나 GC를 기다리기 전에 정리가 끝나야 한다.
    assert closed == ["upstream closed"]


asyncio.run(check())

En una reproducción separada, donde se elimina aclosing(), se observa que la lista de limpieza está vacía al cerrar solo el generador externo, manteniendo la referencia al generador aguas arriba. Al aplicar aclosing(), la limpieza aguas arriba finaliza antes de que regrese response.aclose(). También se comprobó conjuntamente la limpieza y la propagación de la cancelación cuando la tarea de consumo se cancel().

contextlib.aclosing() llama a aclose() de los generadores asíncronos al salir del bloque. La clave es vincular el punto de limpieza con contextos de ejecución como los bucles. Documentación oficial de contextlib de Python

Esta regla debe aplicarse a cada adaptador de transmisión. No solo hay que decidir quién cierra los generadores creados directamente, sino también los flujos devueltos por los SDK. El hecho de que el framework HTTP cierre la respuesta más externa no garantiza el tiempo de vida de todos los recursos aguas arriba.

3. Los errores y la cancelación son flujos de control diferentes

Considere un código que maneja el registro de ejecución de la siguiente manera.

try:
    result = await execute()
except Exception:
    record_failure()
else:
    record_success(result)

En Python 3.12, asyncio.CancelledError es un subtipo de BaseException. Por lo tanto, el except Exception anterior no maneja la cancelación. Al aclose() un generador asíncrono, se introduce GeneratorExit en el punto de suspensión. Esta es la razón para verificar por separado la cancelación y el cierre explícito del generador. Documentación de cancelación de tareas asyncio, Documentación de métodos de generadores asíncronos

El registro de estado de ejecución requiere al menos las siguientes invariantes.

1. Las ejecuciones limpiadas normalmente tienen un estado final.
2. La terminación tardía de la conexión no cambia una ejecución confirmada como exitosa a stopped.
3. Si hay respuestas parciales, se conservan junto con el registro de la detención.
4. Las señales de cancelación se propagan hacia la capa superior una vez finalizada la limpieza.

Dentro de un único ejecutor, variables locales como finalized ayudan a evitar el cierre duplicado. Se cambian a verdadero después de confirmar (commit) el éxito o el fallo, y en la ruta de limpieza final se procesan las ejecuciones que aún no se han cerrado.

Sin embargo, no se debe clasificar todo como cancelación solo por llegar a finally. Los errores de DB en el propio registro de terminación, las interrupciones del proceso y los estados inciertos de los resultados de tareas externas deben tratarse por separado. El alcance que resuelve una única variable local es solo el flujo de control dentro del mismo ejecutor.

4. await dentro de finally también puede cancelarse

Al mover el código de limpieza a finally, se evidencia el siguiente problema.

finally:
    await save_terminal_state()

¿Bajo qué política de cancelación se ejecuta este await?

Dentro de los ámbitos de cancelación de AnyIO, mientras el estado cancelado persista, pueden volver a producirse excepciones de cancelación en cada punto de espera. Por lo tanto, los registros de finalización asíncronos pueden necesitar secciones de limpieza limitadas. Un enfoque es bloquear temporalmente la cancelación del ámbito externo con su propio tiempo límite, como move_on_after(..., shield=True). Esta protección depende de la semántica de los ámbitos de AnyIO y no se debe asumir que bloquea incondicionalmente cualquier asyncio.Task.cancel(). Documentación de cancelación y limpieza de AnyIO

En las secciones de limpieza, conviene distinguir entre:

• Limpieza de recursos: Cerrar sockets, flujos de respuesta y archivos temporales.
• Cierre de estado: Persistir los resultados de la ejecución.
• Marcado para recuperación: Si no se pudo cerrar por tiempo límite o fallo de DB, permitir que la ruta de recuperación lo detecte.

Proteger todo el trabajo de la cancelación invalida las solicitudes de detención del usuario. Por el contrario, si las secciones protegidas son ilimitadas, el procesamiento de terminación no finaliza. El alcance de la protección debe ser breve, y los fallos de cierre deben dejarse como errores observables.

Si se utilizan llamadas a DB sincrónicas, las consultas pueden bloquear el bucle de eventos durante mucho tiempo, incluso si la cancelación await a nivel de Python no interviene. En este caso, también se necesitan tiempos límite de conexión y consulta de la DB, así como políticas de terminación de procesos.

5. El hecho de haber enviado la finalización y el hecho de haberla guardado

La siguiente secuencia crea una ventana de fallo.

완료 이벤트 전송 → DB 커밋

Si el proceso se termina justo después de enviar el evento, el usuario vio la finalización, pero en la pantalla actualizada puede parecer que sigue en ejecución.

Si se debe poder volver a consultar el resultado final, la siguiente secuencia es más fácil de manejar.

결과·종결 상태 저장 → 커밋 → 완료 이벤트 전송

En este orden también puede cortarse la conexión inmediatamente después del commit. Sin embargo, los resultados ya guardados se pueden consultar de nuevo. Se ha desplazado la ventana de fallo hacia una dirección donde la entrega fallida se puede recuperar.

Al usar SQLAlchemy, también se debe distinguir entre flush() e commit(). flush() es la etapa en la que los cambios SQL se reflejan en la transacción actual. No significa que dichos cambios se hayan confirmado (committed) definitivamente. No se debe dar por concluida la verificación del almacenamiento solo porque los objetos actualizados se vean en la misma sesión. Documentación de sesiones y transacciones de SQLAlchemy

La verificación se realiza en una nueva transacción desde una conexión distinta. Aunque la sesión del productor siga abierta, se deben consultar el estado final, la hora de finalización y las respuestas parciales. Las transacciones de lectura abiertas durante mucho tiempo pueden ver instantáneas anteriores según el nivel de aislamiento, por lo que los límites de lectura de las pruebas también deben ser claros.

Tenga cuidado con los diseños que envuelven llamadas externas de larga duración en una sola transacción de DB. Registrar el inicio de la ejecución, los puntos de control de progreso y el cierre en transacciones breves reduce el tiempo de ocupación de conexiones y bloqueos, y facilita establecer criterios de recuperación tras la interrupción del proceso.

6. En múltiples workers, la autoridad de terminación también se debe guardar

Al añadir un ejecutor más, las variables finalized locales no son suficientes.

워커 A: 작업 인수 → 긴 지연
복구기: A의 리스 만료 판단
워커 B: 같은 작업 재인수 → 완료 저장
워커 A: 뒤늦게 복귀 → 오래된 결과 저장 시도

Este problema no se debe resolver mediante "quién escribió por última vez". Solo los intentos con la autoridad de ejecución actual deben poder guardar.

Una opción de diseño es incluir los siguientes valores en el registro de ejecución:

La reasignación de tareas se realiza de forma atómica e incrementa el número de generación. Los ejecutores guardan utilizando el número que recibieron. Por ejemplo, la actualización final tiene la siguiente condición. El SQL es un ejemplo conceptual que utiliza notación de variables de enlace.

UPDATE execution_run
SET state = :terminal_state,
    finished_at = CURRENT_TIMESTAMP,
    result_ref = :result_ref
WHERE run_id = :run_id
  AND execution_epoch = :my_epoch
  AND state = 'running'
  AND lease_until > CURRENT_TIMESTAMP;

Si el número de filas actualizadas es 0, se perdió la autoridad de cierre o el estado ya cambió. Ese ejecutor no debe registrar nuevas notificaciones de finalización ni punteros a resultados posteriores. Las reglas de autoridad se aplican también a los eventos de progreso y puntos de control. Si se guardan archivos primero, conviene utilizar rutas por intento y vincular el puntero al resultado final en la transacción que haya superado la verificación de autoridad.

Utilice el tiempo de la DB de manera consistente para la renovación del lease y la evaluación del vencimiento en la medida de lo posible. Esto reduce los problemas de propiedad de ejecución causados por diferencias de reloj entre procesos. El número de generación actúa como un token de fencing que impide escrituras de ejecutores antiguos.

No obstante, este SQL no garantiza que los efectos secundarios externos se ejecuten solo una vez. Si A tuvo éxito en la solicitud externa pero se detuvo antes de guardar el resultado, el reintento de B podría provocar el mismo efecto. Las claves de idempotencia de las APIs externas, el registro de etapas antes y después de ejecutar el efecto, la consulta de resultados o las operaciones de compensación se deben diseñar por separado. La clave de idempotencia debe separarse del número de generación que cambia en cada reintento, y la misma clave debe utilizarse para el mismo efecto lógico.

7. Las tareas que deben continuar se independizan de la conexión de suscripción

Si se debe seguir ejecutando al cerrar la pestaña, se pueden separar las siguientes tres rutas.

명령: 실행 요청 → 요청 식별자 중복 검사 → 작업 저장
실행: 작업 인수 → 단계 실행 → 상태·결과·이벤트 저장
관측: 저장된 진행 내역 조회 + 실시간 이벤트 구독

Añadir solo llamadas a create_task() puede crear ejecuciones simultáneas dentro del proceso. Pero la recuperación de tareas tras el reinicio del proceso, la asunción de la ejecución y los criterios de reintento no se generan automáticamente.

Al reabrir la suscripción, se debe poder recuperar lo posterior al último evento recibido. SSE tiene el campo estándar id y el comportamiento estándar de enviar Last-Event-ID al reconectar. Pero el almacenamiento y la reproducción de eventos en el servidor deben implementarse en la aplicación. Además, en los clientes de flujo basados en fetch, que no usan el EventSource predeterminado del navegador, la reconexión y la transferencia del cursor deben implementarse manualmente. Estándar de Server-sent events de WHATWG

Si se diseñan eventos persistentes, (run_id, sequence) puede servir para eliminar duplicados y restaurar el orden dentro de la misma ejecución. Guardar también el número de intento facilita distinguir los registros antes y después de los reintentos. Sin embargo, asignar números crecientes no garantiza el orden de commit. Si hay varios escritores, se necesitan reglas de cursor que consideren la serialización por ejecución o el orden de commit.

Tampoco es imprescindible guardar todos los tokens como eventos persistentes. Se pueden guardar el inicio, el fin y el resultado de cada etapa, y consolidar los tokens en bloques de cierto tamaño o instantáneas periódicas. La precisión de reproducción necesaria y el coste de escritura se deben decidir conjuntamente.

8. Las notificaciones de finalización se derivan del estado guardado

Entre el guardado del estado de finalización y el envío de la notificación también existe una ventana de fallo.

• Fallo después de guardar el estado y antes de enviar la notificación: Terminó, pero no hay notificación.
• Fallo después de enviar la notificación y antes de guardar la confirmación de entrega: Se reintenta y la misma notificación se repite.

Una opción es guardar también un registro para el envío de notificaciones dentro de la transacción que cambia el estado final. Un remitente independiente lee este registro, envía la notificación y registra el resultado de la entrega. Los registros de notificación solo deben generarse cuando la actualización de cierre supera las condiciones de propiedad anteriores.

Dado que el envío en sí puede repetirse, se debe asignar un event_id estable a las notificaciones y eliminar los duplicados en el lado receptor. Las restricciones que hacen únicas las notificaciones de cierre de la misma ejecución también son útiles. Este método preserva conjuntamente el estado y la intención de notificación, pero no garantiza una entrega exactamente-una-vez en los canales de notificación externos.

La redacción de las notificaciones en pantalla también debe reflejar este contrato. La notificación de "finalización" debe permitir abrir el resultado guardado, y "detenido" no debe interpretarse como que se han cancelado todos los cambios en los sistemas externos. Los efectos ya producidos mientras la tarea estaba en progreso y la interrupción de la generación de la respuesta son hechos distintos.

9. Pruebe los límites de terminación antes que las respuestas normales

Si se define la unidad de verificación en el HTTP 200, es fácil pasar por alto estos problemas. Es más eficaz especificar, como pares, la posición donde se inyecta el fallo y el estado que debe cumplirse posteriormente.

En las reproducciones mínimas de este artículo se comprobó directamente el cierre justo después del primer trozo, la limpieza inmediata mediante aclosing() y la propagación de la cancelación de tareas durante la espera. En las verificaciones de integración independientes con DB, se comprobó mediante conexiones independientes, inmediatamente después de cerrar el flujo, que el estado de detención, las respuestas parciales y la hora de finalización se habían confirmado. Las entradas de reasignación del lease y reenvío de notificaciones son escenarios de inyección de fallos que deben añadirse al implementar los diseños propuestos.

Las métricas operativas tampoco bastan con el tiempo medio de respuesta. Valores como el número de ejecuciones con leases caducados, el número de fallos de registro de cierre, la diferencia de tiempo entre el estado final y la entrega de la notificación, y las veces que se rechazaron escrituras de generaciones antiguas muestran la salud de la ruta de recuperación.

Conclusión

En los sistemas de transmisión, la terminación consta de varias etapas: detener el consumo, limpiar los recursos aguas arriba, confirmar el estado de ejecución y transmitir ese hecho al usuario.

En las revisiones de diseño, vale la pena verificar primero estas tres preguntas.

1. ¿Quién tiene actualmente la responsabilidad de terminar esta tarea?
2. Si ese ejecutor desaparece, ¿queda un registro que permita al siguiente ejecutor tomar una decisión?
3. Si se consulta de nuevo el estado final enviado al usuario, ¿se obtiene el mismo resultado?

Si estos tres puntos son claros, la cancelación no queda oculta en una línea de manejo de excepciones. Se convierte en un protocolo de ejecución que puede verificarse al mismo nivel que el éxito normal.