AsyncPostgresStore cleanup leaves pending background batch tasks causing "Task was destroyed" warnings

Checked other resources

• This is a bug, not a usage question. For questions, please use the LangChain Forum (https://forum.langchain.com/).
• I added a clear and detailed title that summarizes the issue.
• I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
• I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

import asyncio
from langgraph.store.postgres.aio import AsyncPostgresStore

async def reproduce_bug():
    """
    Minimal reproducible example demonstrating the background task cleanup issue.
    """
    # PostgreSQL connection string
    conn_string = "postgresql://user:password@localhost:5432/database"

    # Embedding configuration (using OpenAI embeddings)
    index_config = {
        "dims": 1536,
        "embed": "openai:text-embedding-3-small"
    }

    # Use AsyncPostgresStore as context manager
    async with AsyncPostgresStore.from_conn_string(
        conn_string,
        index=index_config
    ) as store:
        # Setup the store (creates tables if needed)
        await store.setup()

        # Perform some store operations
        namespace = ("test_namespace",)
        await store.put(namespace, "key1", {"value": "test_data_1"})
        await store.put(namespace, "key2", {"value": "test_data_2"})
        await store.put(namespace, "key3", {"value": "test_data_3"})

        # Read back
        items = await store.search(namespace)
        print(f"Stored {len(items)} items")

    # After exiting the context manager, background batch tasks are still pending
    # This causes asyncio warnings when the event loop closes

if __name__ == "__main__":
    asyncio.run(reproduce_bug())

Error Message and Stack Trace (if applicable)

[ERROR] asyncio: Task was destroyed but it is pending!
task: <Task cancelling name='Task-56124' coro=<_run() running at
/opt/venv/lib/python3.12/site-packages/langgraph/store/base/batch.py:330>
wait_for=<Future cancelled>>

[ERROR] asyncio: Task was destroyed but it is pending!
task: <Task cancelling name='Task-56692' coro=<_run() running at
/opt/venv/lib/python3.12/site-packages/langgraph/store/base/batch.py:330>
wait_for=<Future cancelled>>

Description

Note: The error appears when the Python script/application exits, not during the async with block itself.

What I'm doing:

I'm using AsyncPostgresStore with the context manager pattern (async with) in a FastAPI application that handles SSE (Server-Sent Events) streaming. The store operations work correctly and data is saved properly.

What I expect to happen:

When exiting the async with AsyncPostgresStore.from_conn_string(...) as store: block, the context manager's __aexit__ method should properly clean up all resources, including:

• Waiting for all background batch processing tasks to complete
• Closing database connections
• No asyncio warnings or errors

What actually happens:

The __aexit__ method closes the database connection but does not wait for background batch processing tasks (created in langgraph/store/base/batch.py) to complete. This results in:

1. Background tasks remain in pending state
2. When the event loop/application exits, asyncio detects these orphaned tasks
3. Error logs: Task was destroyed but it is pending!

Root Cause Analysis:

LangGraph's AsyncPostgresStore uses background asyncio tasks for batch processing optimization (in langgraph/store/base/batch.py line 330). However, the context manager cleanup (__aexit__) does not properly await these tasks before closing:

# Expected behavior in __aexit__:
async def __aexit__(self, ...):
    await self._wait_for_batch_tasks()  # Missing!
    await self.conn.close()

# Actual behavior:
async def __aexit__(self, ...):
    await self.conn.close()  # Closes without waiting for background tasks

Impact:

• ✅ Data integrity: Not affected (data is saved correctly)
• ⚠️ Resource cleanup: Background tasks are not properly terminated
• ⚠️ Logging: Error messages pollute logs
• ⚠️ Best practices: Violates Python context manager protocol

Workaround:

We currently work around this by adding a small delay before exiting the context manager:

async with AsyncPostgresStore.from_conn_string(...) as store:
    yield store, checkpointer

    # Workaround: Wait for background batch tasks to complete
    await asyncio.sleep(0.15)  # 150ms delay

This is not ideal and should be fixed in LangGraph itself.

System Info

System Information
------------------
> OS:  Linux
> OS Version:  #1 SMP PREEMPT_DYNAMIC Thu Jun  5 18:30:46 UTC 2025
> Python Version:  3.12.12 (main, Oct 21 2025, 02:11:48) [GCC 12.2.0]

Package Information
-------------------
> langchain_core: 1.0.2
> langchain: 1.0.3
> langchain_community: 1.0.0a1
> langsmith: 0.4.38
> langgraph: 1.0.2
> langgraph-checkpoint: 3.0.0
> langgraph-checkpoint-postgres: 3.0.0

Other Dependencies
------------------
> asyncpg: 0.30.0
> psycopg: 3.2.12
> psycopg-pool: 3.2.7
> SQLAlchemy: 2.0.44
> pydantic: 2.12.3

Start with AsyncPostgresStore's context-manager cleanup and inspect the batch processing task shown in langgraph/store/base/batch.py around line 330. Run the minimal AsyncPostgresStore example against PostgreSQL and observe shutdown behavior. Done means exiting the async context waits for background batch tasks and produces no pending-task warnings.