mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-28 10:49:56 +02:00
6139dacef9
### Summary This PR fixes an issue where `AsyncPregelLoop` could leave behind an orphaned `stream.wait()` task, resulting in warnings like: ``` Task was destroyed but it is pending! ``` ### Related Discussion This PR is in response to: [langchain-ai/langgraph#6163](https://github.com/langchain-ai/langgraph/discussions/6163) ### Problem * In the async path, `get_waiter()` was creating a new `asyncio.Task` via ```python aioloop.create_task(stream.wait()) ``` but never tracked or cleaned it up. * On cancellation or shutdown, these tasks remained pending and produced warnings. ### Solution * Changed `get_waiter()` to: * Maintain a **single waiter task** (similar to the sync path). * Auto-clear the reference when the task finishes. * Added `_cleanup_waiter()`: * On exit, attempt to wake the waiter (`stream._count.release()` if available). * Otherwise, cancel and `await` the pending task to ensure proper cleanup. * Wrapped the `while loop.tick():` block in a `try/finally` to guarantee `_cleanup_waiter()` runs on exit. * Added missing `import contextlib`. ### Impact * Prevents orphaned `stream.wait()` tasks. * Removes noisy `"Task was destroyed but it is pending!"` warnings. * Behavior of async streaming remains unchanged, only lifecycle management improved. ### Test Plan * Reproduced the issue by running async streaming with cancellation. * Verified warnings no longer appear after the fix. * Ran existing test suite (all passing). ### Notes * Sync and Async implementations now follow the same principle: *only one waiter at a time, always cleaned up on exit*. * Backwards-compatible; no API changes. ### Repro & Verification To confirm the issue and the fix I used the following minimal repro snippet: ```python # lg_repro.py import asyncio import os # Enable asyncio debug logs to surface pending task warnings os.environ.setdefault("PYTHONASYNCIODEBUG", "1") from langgraph.graph import START, END, StateGraph State = dict # Slow async node: processes once, then sleeps to keep the waiter alive async def slow_node(state: State) -> State: await asyncio.sleep(0.2) # simulate work state["count"] = state.get("count", 0) + 1 await asyncio.sleep(1.0) # keep stream.wait() waiter active return state # Build simple graph: START -> slow_node -> END builder = StateGraph(State) builder.add_node("slow", slow_node) builder.add_edge(START, "slow") builder.add_edge("slow", END) graph = builder.compile() async def run_and_cancel(): # astream with messages mode triggers internal stream.wait() waiter async def consumer(): async for _ in graph.astream({"msg": "hi"}, stream_mode="messages"): await asyncio.sleep(0.05) t = asyncio.create_task(consumer(), name="astream-consumer") # Allow the stream to start, then cancel the consumer await asyncio.sleep(0.1) t.cancel() try: await t except asyncio.CancelledError: pass # Let loop settle to show pending waiter task if not cleaned await asyncio.sleep(0.05) def main(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.set_debug(True) try: loop.run_until_complete(run_and_cancel()) finally: # If the internal waiter is not cleaned, closing the loop will warn loop.close() if __name__ == "__main__": main() ```` **How to run** ```powershell # Before (main branch) git checkout main pip install -e libs/langgraph $env:PYTHONASYNCIODEBUG=1; python lg_repro.py # After (patched branch) git checkout async-waiter-cleanup pip install -e libs/langgraph $env:PYTHONASYNCIODEBUG=1; python lg_repro.py ``` **Observed results** * **main branch (before fix):** Shows warnings like: ``` Task was destroyed but it is pending! ... coro=<AsyncQueue.wait() ...> created at langgraph/pregel/main.py:2927 ``` * **patched branch (after fix):** No warnings. The single waiter is properly cleaned up on exit via `_cleanup_waiter()` (release semaphore if available, then cancel/await). --- This confirms that the patch removes the orphaned `stream.wait()` task and prevents `"Task was destroyed but it is pending!"` warnings during cancellation/shutdown. --------- Co-authored-by: Caspar Broekhuizen <caspar@langchain.dev>