langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

[PR #6167] fix(langgraph): cleanup orphaned waiter task in AsyncPregelLoop #4878

New Issue
Closed
opened 2026-02-20 17:50:53 -05:00 by yindo · 0 comments
yindo commented 2026-02-20 17:50:53 -05:00
Owner
Copy Link
Copy Source

Original Pull Request: https://github.com/langchain-ai/langgraph/pull/6167

State: closed
Merged: Yes

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

Problem

  • In the async path, get_waiter() was creating a new asyncio.Task via

    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:

# 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

# 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.

**Original Pull Request:** https://github.com/langchain-ai/langgraph/pull/6167 **State:** closed **Merged:** Yes --- ### 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.
yindo added the pull-request label 2026-02-20 17:50:53 -05:00
yindo closed this issue 2026-02-20 17:50:53 -05:00
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label pull-request
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#4878
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?