Designing Graceful Termination for Streaming Executions: Understanding the Running State That Remains After a Connection Drops

Streaming APIs seem simple on the happy path: start an external operation, transmit result fragments, and change the execution status to succeeded at the end.

But the story changes if a user closes their tab in the middle of a response. The HTTP connection terminates, yet the model call continues, or the generator closes while the execution record remains running. The screen that received the completion event and the final state in the database may differ.

To resolve this, you must define the lifetimes of the connection, execution, and persistent record separately, and link their termination responsibilities. This article starts from a minimal reproduction in Python async generators, extending to cancellation handling, transaction boundaries, and execution ownership across multiple workers.

Examples were verified on Python 3.12.13. Explanations regarding AnyIO and SQLAlchemy are based on the 4.x and 2.0 documentation, respectively. The job lease and event storage structures in later sections are design examples, distinct from the reproduction experiments in earlier sections.

1. First, Separate the Four Lifetimes

These four lifetimes do not always need to align. Instead, you must define linking rules based on product behavior.

• Execution dependent on the request: Treat connection breakage as a cancellation signal. Cleanup responsibility lies with the executor handling the request.
• Tasks that must continue: Treat connection breakage as a subscription unsubscription. Execution responsibility is held by an executor independent of the request.

If judgment here is unclear, contradictory code emerges within the same program. One layer cancels tasks upon connection closure, while another waits for completion notifications. Both are individually natural, but the contract of the integrated system fails.

2. Exiting async for Does Not Immediately Close the Upstream Generator

Nested streaming is typically structured as follows:

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

If each layer simply reads via async for and only handles yield, the termination of the outer generator may not trigger explicit termination of the inner generator. This is particularly critical when the outer layer stops immediately after passing a value.

The following example verifies the timing of upstream resource cleanup.

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())

In a separate reproduction where aclosing() was removed, closing only the outer generator while maintaining a reference to the upstream generator resulted in an empty cleanup list. Applying aclosing() ensured upstream cleanup completed before response.aclose() returned. We also verified cleanup and cancellation propagation when the consuming task was cancel().

contextlib.aclosing() calls the async generator’s aclose() upon exiting the block. The key point is binding cleanup timing to the same execution context as the loop. Python contextlib official documentation

This rule must be applied to each streaming adapter. You must define who closes not only generators you built directly but also streams returned by SDKs. The fact that the HTTP framework closes the outermost response does not guarantee the lifetime of all upstream resources.

3. Errors and Cancellations Are Different Control Flows

Consider code that handles execution records as follows:

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

In Python 3.12, asyncio.CancelledError is a subclass of BaseException. Therefore, the above except Exception does not handle cancellation. When an async generator is aclose(), GeneratorExit is raised at the stopped point. This is why cancellation and explicit generator closure must be verified separately. asyncio task cancellation documentation, async generator method documentation

Execution status records require at least the following invariants.

1. Properly cleaned-up executions have a final status.
2. A late connection termination does not change an execution finalized as successful to stopped.
3. If there is a partial response, it is preserved along with a stop record.
4. Cancellation signals are propagated to the upper layer after cleanup is complete.

Within a single executor, local variables like finalized help prevent duplicate closure. Set this to true after committing success or failure, and handle any execution not yet finalized in the last cleanup path.

However, you should not classify every case as cancellation merely because finally was reached. DB errors in the termination record itself, process interruptions, and unclear outcomes of external operations must be handled separately. The scope solvable by a single local variable is limited to control flow within the same executor.

4. await Inside finally Can Also Be Canceled

Moving cleanup code to finally reveals the following issue.

finally:
    await save_terminal_state()

Under what cancellation policy is this await executed?

Inside an AnyIO cancellation scope, cancellation exceptions may reoccur at each await point while the cancellation state persists. Thus, asynchronous termination records may need a restricted cleanup section. This involves briefly blocking cancellation from the outer scope, as in move_on_after(..., shield=True), but with its own timeout. This protection depends on the semantics of the AnyIO scope; it should not be assumed to unconditionally block until an arbitrary asyncio.Task.cancel(). AnyIO cancellation & cleanup documentation

In cleanup sections, it is better to distinguish:

• Resource cleanup: Close sockets, response streams, and temporary files.
• State finalization: Persist execution results.
• Marking for recovery: If finalization failed due to timeout or DB failure, ensure the recovery path can discover it.

Protecting the entire task from cancellation renders the user’s stop request ineffective. Conversely, unlimited protection prevents termination processing from completing. Keep protection scopes short and leave finalization failures as observable errors.

If using synchronous DB calls, even if Python-level await cancellation does not interrupt, queries may block the event loop for a long time. In this case, DB connection/query timeouts and process termination policies are also necessary.

5. The Fact of Sending Completion vs. The Fact of Saving Completion

The following sequence creates a failure window.

완료 이벤트 전송 → DB 커밋

If the process terminates immediately after sending the event, the user sees completion, but a refreshed screen may show it as in-progress.

If final results must be retrievable, the following sequence is easier to handle.

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

In this order, the connection may drop immediately after commit. However, the saved result can be re-queried. This shifted the failure window to a direction where delivery failures can be recovered.

When using SQLAlchemy, you must also distinguish between flush() and commit(). flush() is the step where change SQL is reflected in the current transaction. It does not mean the change has been finally committed. Do not consider storage verification complete just because updated objects are visible in the same session. SQLAlchemy session & transaction documentation

Verification is performed in a new transaction on a different connection. Even if the producer session is still open, the final status, end time, and partial response must be retrievable. Long-open read transactions may see previous snapshots depending on isolation levels, so the read boundary in tests must also be clear.

Be cautious of designs wrapping long external calls in a single DB transaction. Recording execution start, progress checkpoints, and termination in short transactions reduces connection and lock hold times and facilitates recovery criteria after process interruption.

6. In Multi-Worker Setups, Termination Rights Must Be Stored

Adding another executor makes local finalized variables insufficient.

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

This problem should not be resolved by “who wrote last.” Only the attempt holding current execution rights must be able to save.

One design places the following values in the execution record.

Job reclaims are performed atomically, incrementing the generation number. Executors save using the number they received. For example, final updates have the following conditions. SQL is a conceptual example using bind variable notation.

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;

If the update count is 0, finalization rights were lost or the status had already changed. That executor must not record new finalization notifications or subsequent result pointers. The same rights rule applies to progress events and checkpoints. If files are saved first, use per-attempt paths, and connect the final result pointer in the transaction that successfully verified rights.

Use DB time consistently for lease renewals and expiration judgments to minimize issues arising from per-process clock differences affecting execution ownership. Generation numbers act as fencing tokens to prevent writes from older executors.

However, this SQL does not guarantee that external side effects are executed exactly once. If A succeeds in an external request but stops before saving the result, B’s retry may cause the same effect. Idempotency keys for external APIs, step recording before and after effect execution, and result queries or compensation operations must be designed separately. Idempotency keys should be separated from generation numbers that change per retry; the same key must be used for the same logical effect.

7. Decouple Tasks That Must Continue from the Subscription Connection

If execution must continue even when the tab is closed, you can separate the following three paths.

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

Adding only create_task() calls creates concurrent execution within the process. However, job recovery after process restart, execution claiming, and retry criteria do not automatically appear.

When reopening a subscription, you must be able to restore events since the last received one. SSE has standard behavior using the id field and sending Last-Event-ID upon reconnection. However, event storage and replay on the server must be implemented by the application. Additionally, stream clients based on fetch, rather than the browser’s default EventSource, must implement reconnection and cursor delivery directly. WHATWG Server-sent events standard

If designing persistent events, you can deduplicate based on (run_id, sequence) and restore order within the same execution. Storing attempt numbers alongside helps distinguish records before and after retries. However, assigning incrementing numbers does not guarantee commit order. With multiple writers, cursor rules considering per-execution serialization or commit order are necessary.

Persisting all tokens as events is not mandatory. You can choose to store step starts, ends, and results, while aggregating tokens into chunks or periodic snapshots of a certain size. Determine the necessary replay accuracy and write costs together.

8. Derive Completion Notifications from the Saved State

There is also a failure window between saving the completion status and sending the notification.

• Failure after status save, before notification send: Completed, but no notification.
• Failure after notification send, before delivery confirmation save: Same notification repeats during retries.

One method is to save the notification dispatch record within the transaction that changes the final status. A separate dispatcher reads this record, sends the notification, and records delivery results. Notification records should only be created if the final update passed the aforementioned ownership conditions.

Since transmission itself can repeat, notifications must have a stable event_id, and recipients should deduplicate. Constraints making the final notification for the same execution unique are also helpful. This approach preserves state and notification intent together but does not guarantee exactly-once delivery to external notification channels.

On-screen notification wording must reflect this contract. A “completed” notification must allow opening the saved result, and “stopped” should not imply that all changes in external systems were canceled. Effects already occurred during job progress and interruption of response generation are different facts.

9. Test Termination Boundaries More Than Happy Paths

If validation units are set to HTTP 200, it is easy to miss these issues. It is more effective to document pairs of failure injection locations and the state that must hold afterward, as shown below.

In this article’s minimal reproduction, we directly verified closing immediately after the first chunk, immediate cleanup via aclosing(), and propagation of task cancellation during waiting. In separate integration tests using a DB, we verified via an independent connection immediately after stream closure that stopped status, partial response, and end time were committed. Lease reclaim and notification retransmission items are failure injection scenarios to add when implementing the proposed design in practice.

Operational metrics are also insufficient with average response time alone. Values like the number of executions with expired leases, finalization record failures, time difference between final status and notification delivery, and rejection counts for writes from old execution generations indicate the health of the recovery path.

Conclusion

Termination in streaming systems consists of multiple steps: stopping consumption, cleaning up upstream resources, finalizing execution status, and conveying that fact to the user.

In design reviews, it is worth first checking the following three questions.

1. Who currently bears the responsibility for ending this job?
2. If that executor disappears, does a record remain for the next executor to make decisions?
3. If you re-query the final status sent to the user, do you get the same result?

If these three are clear, cancellation is not buried in a single line of exception handling. It becomes an execution protocol that can be verified at the same level as normal completion.