DB connection leak

[yangxikun]

Bug Description

DB connection pool overflow:

Reproduction

1. Create simple flow

2. curl call api: api/v1/build/47b14ff4-f165-44c5-af90-d6a742ff7f6f/flow

curl --location 'http://localhost:3000/api/v1/build/47b14ff4-f165-44c5-af90-d6a742ff7f6f/flow' \
--header 'x-api-key: sk-xxx-xxx' \
--header 'Content-Type: application/json' \
--data '{
    "inputs": {
        "input_value": "hello",
        "session": "xxx"
    }
}'

Expected behavior

Release unused db connection.

Who can help?

@ogabrielluiz

Operating System

mac

Langflow Version

v1.1.1

Python Version

3.12

Screenshot

Every curl will leak a db connection:

I debugged and found the problem is in build_flow->event_generator:

If not data, build_graph_and_get_order will run by asyncio.create_task, this will cause AsyncDbSession doesn't close.

If run build_graph_and_get_order directly with await, AsyncDbSession close correctly.

Flow File

No response

[dosubot]

The issue you're encountering with the database connection leak seems to be related to how the build_graph_and_get_order function is being executed. When using asyncio.create_task, the AsyncDbSession might not be closing properly, leading to a connection pool overflow.

To address this, you can try running build_graph_and_get_order directly with await instead of using asyncio.create_task. This change ensures that the AsyncDbSession is properly managed and closed, preventing the connection leak.

Here's a potential modification to the event_generator function:

async def event_generator(event_manager: EventManager, client_consumed_queue: asyncio.Queue) -> None:
    if not data:
        try:
            ids, vertices_to_run, graph = await build_graph_and_get_order()
        except asyncio.CancelledError:
            return
        except Exception as e:
            error_message = ErrorMessage(
                flow_id=flow_id,
                exception=e,
            )
            event_manager.on_error(data=error_message.data)
            raise
    else:
        try:
            ids, vertices_to_run, graph = await build_graph_and_get_order()
        except Exception as e:
            error_message = ErrorMessage(
                flow_id=flow_id,
                exception=e,
            )
            event_manager.on_error(data=error_message.data)
            raise

    event_manager.on_vertices_sorted(data={"ids": ids, "to_run": vertices_to_run})
    await client_consumed_queue.get()

    tasks = []
    for vertex_id in ids:
        task = asyncio.create_task(build_vertices(vertex_id, graph, client_consumed_queue, event_manager))
        tasks.append(task)
    try:
        await asyncio.gather(*tasks)
    except asyncio.CancelledError:
        for task in tasks:
            task.cancel()
        return
    except Exception as e:
        logger.error(f"Error building vertices: {e}")
        error_message = ErrorMessage(
            flow_id=flow_id,
            exception=e,
        )
        event_manager.on_error(data=error_message.data)
        raise
    event_manager.on_end(data={})
    await event_manager.queue.put((None, None, time.time))

By using await directly, you ensure that the session lifecycle is properly managed, which should help in releasing unused database connections.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[severfire]

@yangxikun is it for both sqllite and postgresql databases? I found out that there is somewhere leak in memory, while i run like 25k api calls, the RAM usage can go even to 10GB or more.

may be related:
#5287
#5275
#4008

@Cristhianzl

[yangxikun]

@severfire I just test postgresql

[Cristhianzl]

@severfire we are already working to improve It!
thanks for your feedback!

[severfire]

@Cristhianzl thank you! please let me know how I can help here. In general I am using sqlite, as I do not store really data in langflow. Just using it to analyze what I need via API and I use simple installation with python or uv. For me the goal is to have like up to 500k API calls a month cause my client seems to grow. I love langflow and I would love to use it, yet I am worried it could be optimized bit better :-) for example if the worker could be lighter regarding RAM usage, memory leaks etc... I hate to complain and so on, yet I really want to consider langflow for business use :-)

[cbornet]

Well, although it is not really useful to create a task and to explicitly propagate the cancellation (cancelations are already propagated by coroutines awaiting other coroutines), I don't see why it makes a difference for the AsyncDbSession.
So we can make the change but it would be nice to understand why it fixes the issue.
Also it seems the behavior is identical whether data is None or not so the if loop doesn't seem needed.

[severfire]

@cbornet Hello, Were you able to see #5275 ? Hmm... Do you see RAM memory increase when doing a lot of API calls? :-/ Thank you for your hard work!

[cbornet]

Do you see RAM memory increase when doing a lot of API calls?

We have done some fixes for the memory leaks. It seems to be a different issue there with DB connections.

[cbornet]

@yangxikun I couldn't reproduce the issue on latest main branch. I tested both sqlite and postgres.
Can you check if you still have the issue on latest main ?