From f30b291f80ae4b20c30dcf100b8a57bd6717466f Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 7 Apr 2026 16:12:19 -0700 Subject: [PATCH] fix: Replace aiofile with aiofiles to prevent caio context leak under concurrent execution (#12525) * fix: replace aiofile with aiofiles to prevent caio context leak under concurrent execution aiofile uses caio (kernel AIO) which creates contexts in a global dict that are never cleaned up. Under concurrent execution these accumulate until the OS aio-max-nr limit is exhausted, causing SystemError(11, 'Resource temporarily unavailable'). aiofiles uses thread pools instead and does not have this issue. Migrates all aiofile.async_open usages across both backend and lfx packages to aiofiles.open. Based on #12433 by @manav2000, extended to cover all remaining usages. Co-Authored-By: manav2000 * test: add concurrent write-then-read regression test for caio EAGAIN fix Exercises the exact failure pattern from #12414: multiple concurrent save-then-immediately-read operations on the storage service. This would previously trigger SystemError(11, EAGAIN) after ~150-200 runs with the aiofile/caio backend. Co-Authored-By: manav2000 * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) --------- Co-authored-by: manav2000 Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- .../base/langflow/api/v1/flows_helpers.py | 5 ++-- .../base/langflow/initial_setup/setup.py | 8 +++--- .../langflow/services/flow/flow_runner.py | 4 +-- .../base/langflow/services/storage/local.py | 8 +++--- .../tests/unit/api/v1/test_flows_helpers.py | 2 +- .../storage/test_local_storage_service.py | 25 +++++++++++++++++++ .../directory_reader/directory_reader.py | 6 ++--- src/lfx/src/lfx/load/load.py | 6 ++--- src/lfx/src/lfx/services/settings/base.py | 4 +-- 9 files changed, 46 insertions(+), 22 deletions(-) diff --git a/src/backend/base/langflow/api/v1/flows_helpers.py b/src/backend/base/langflow/api/v1/flows_helpers.py index 2be5eb8ab4..14c3b38767 100644 --- a/src/backend/base/langflow/api/v1/flows_helpers.py +++ b/src/backend/base/langflow/api/v1/flows_helpers.py @@ -13,7 +13,7 @@ from pathlib import Path as StdlibPath from typing import TYPE_CHECKING, Any from uuid import UUID -from aiofile import async_open +import aiofiles from anyio import Path from fastapi import HTTPException from fastapi.responses import StreamingResponse @@ -176,8 +176,7 @@ async def _save_flow_to_fs(flow: Flow, user_id: UUID, storage_service: StorageSe try: safe_path = _get_safe_flow_path(flow.fs_path, user_id, storage_service) await safe_path.parent.mkdir(parents=True, exist_ok=True) - # async_open expects a string path, not a Path object - async with async_open(str(safe_path), "w") as f: + async with aiofiles.open(str(safe_path), "w") as f: await f.write(flow.model_dump_json()) except HTTPException: raise diff --git a/src/backend/base/langflow/initial_setup/setup.py b/src/backend/base/langflow/initial_setup/setup.py index dab15402b8..e749cdc102 100644 --- a/src/backend/base/langflow/initial_setup/setup.py +++ b/src/backend/base/langflow/initial_setup/setup.py @@ -13,11 +13,11 @@ from tempfile import TemporaryDirectory from typing import AnyStr from uuid import UUID +import aiofiles import anyio import httpx import orjson import sqlalchemy as sa -from aiofile import async_open from emoji import demojize, purely_emoji from lfx.base.constants import ( FIELD_FORMAT_ATTRIBUTES, @@ -676,7 +676,7 @@ def get_project_data(project): async def update_project_file(project_path: anyio.Path, project: dict, updated_project_data) -> None: project["data"] = updated_project_data - async with async_open(str(project_path), "w", encoding="utf-8") as f: + async with aiofiles.open(str(project_path), "w", encoding="utf-8") as f: await f.write(orjson.dumps(project, option=ORJSON_OPTIONS).decode()) await logger.adebug(f"Updated starter project {project['name']} file") @@ -803,7 +803,7 @@ async def load_agentic_flows() -> list[tuple[anyio.Path, dict]]: await logger.adebug("Loading agentic flows") async for file in folder.glob("*.json"): try: - async with async_open(str(file), "r", encoding="utf-8") as f: + async with aiofiles.open(str(file), encoding="utf-8") as f: content = await f.read() flow = orjson.loads(content) agentic_flows.append((file, flow)) @@ -961,7 +961,7 @@ async def load_flows_from_directory() -> None: if not await anyio.Path(file_path).is_file() or file_path.suffix != ".json": continue await logger.ainfo(f"Loading flow from file: {file_path.name}") - async with async_open(str(file_path), "r", encoding="utf-8") as f: + async with aiofiles.open(str(file_path), encoding="utf-8") as f: content = await f.read() await upsert_flow_from_file(content, file_path.stem, session, user.id) diff --git a/src/backend/base/langflow/services/flow/flow_runner.py b/src/backend/base/langflow/services/flow/flow_runner.py index d4d67529be..fc9c870db6 100644 --- a/src/backend/base/langflow/services/flow/flow_runner.py +++ b/src/backend/base/langflow/services/flow/flow_runner.py @@ -3,7 +3,7 @@ import os from pathlib import Path from uuid import UUID, uuid4 -from aiofile import async_open +import aiofiles from lfx.graph import Graph from lfx.graph.vertex.param_handler import ParameterHandler from lfx.log.logger import configure, logger @@ -254,7 +254,7 @@ class LangflowRunnerExperimental: @staticmethod async def get_flow_dict(flow: Path | str | dict) -> dict: if isinstance(flow, str | Path): - async with async_open(Path(flow), encoding="utf-8") as f: + async with aiofiles.open(Path(flow), encoding="utf-8") as f: content = await f.read() return json.loads(content) # If input is a dictionary, assume it's a JSON object diff --git a/src/backend/base/langflow/services/storage/local.py b/src/backend/base/langflow/services/storage/local.py index a2f71b0603..d9e33e179c 100644 --- a/src/backend/base/langflow/services/storage/local.py +++ b/src/backend/base/langflow/services/storage/local.py @@ -5,7 +5,7 @@ from __future__ import annotations from pathlib import Path from typing import TYPE_CHECKING -from aiofile import async_open +import aiofiles from langflow.logging.logger import logger from langflow.services.storage.service import StorageService @@ -169,7 +169,7 @@ class LocalStorageService(StorageService): try: mode = "ab" if append else "wb" - async with async_open(str(file_path), mode) as f: + async with aiofiles.open(str(file_path), mode) as f: await f.write(data) action = "appended to" if append else "saved" await logger.ainfo(f"File {file_name} {action} successfully in flow {flow_id}.") @@ -196,7 +196,7 @@ class LocalStorageService(StorageService): msg = f"File {file_name} not found in flow {flow_id}" raise FileNotFoundError(msg) - async with async_open(str(file_path), "rb") as f: + async with aiofiles.open(str(file_path), "rb") as f: content = await f.read() logger.debug(f"File {file_name} retrieved successfully from flow {flow_id}.") @@ -210,7 +210,7 @@ class LocalStorageService(StorageService): msg = f"File {file_name} not found in flow {flow_id}" raise FileNotFoundError(msg) - async with async_open(str(file_path), "rb") as f: + async with aiofiles.open(str(file_path), "rb") as f: while True: chunk = await f.read(chunk_size) if not chunk: diff --git a/src/backend/tests/unit/api/v1/test_flows_helpers.py b/src/backend/tests/unit/api/v1/test_flows_helpers.py index 045a3feb14..819fd98b0b 100644 --- a/src/backend/tests/unit/api/v1/test_flows_helpers.py +++ b/src/backend/tests/unit/api/v1/test_flows_helpers.py @@ -72,7 +72,7 @@ async def test_save_flow_to_fs_returns_500_on_os_error(current_user, storage_ser ) with ( - patch("langflow.api.v1.flows_helpers.async_open", side_effect=OSError("disk full")), + patch("langflow.api.v1.flows_helpers.aiofiles.open", side_effect=OSError("disk full")), pytest.raises(HTTPException) as exc_info, ): await _save_flow_to_fs(flow, current_user.id, storage_service) diff --git a/src/backend/tests/unit/services/storage/test_local_storage_service.py b/src/backend/tests/unit/services/storage/test_local_storage_service.py index b9e15ec0aa..bd4a4fafa4 100644 --- a/src/backend/tests/unit/services/storage/test_local_storage_service.py +++ b/src/backend/tests/unit/services/storage/test_local_storage_service.py @@ -361,3 +361,28 @@ class TestLocalStorageServiceEdgeCases: # Verify all files were saved listed = await local_storage_service.list_files(flow_id) assert len(listed) == 10 + + async def test_concurrent_write_then_read(self, local_storage_service): + """Regression test for SystemError(11, 'Resource temporarily unavailable'). + + Under concurrent execution, aiofile/caio would leak kernel AIO contexts + causing EAGAIN after ~150-200 runs. This test verifies that writing a file + and immediately reading it back works reliably under concurrency with the + aiofiles backend. See https://github.com/langflow-ai/langflow/issues/12414 + """ + flow_id = "concurrent_rw_flow" + num_files = 50 + + async def write_then_read(i: int) -> None: + file_name = f"file_{i}.bin" + data = f"payload-{i}".encode() + await local_storage_service.save_file(flow_id, file_name, data) + retrieved = await local_storage_service.get_file(flow_id, file_name) + assert retrieved == data, f"file_{i} content mismatch" + + async with anyio.create_task_group() as tg: + for i in range(num_files): + tg.start_soon(write_then_read, i) + + listed = await local_storage_service.list_files(flow_id) + assert len(listed) == num_files diff --git a/src/lfx/src/lfx/custom/directory_reader/directory_reader.py b/src/lfx/src/lfx/custom/directory_reader/directory_reader.py index cf0a84f4dc..22d25cbe5d 100644 --- a/src/lfx/src/lfx/custom/directory_reader/directory_reader.py +++ b/src/lfx/src/lfx/custom/directory_reader/directory_reader.py @@ -3,8 +3,8 @@ import asyncio import zlib from pathlib import Path +import aiofiles import anyio -from aiofile import async_open from lfx.custom.custom_component.component import Component from lfx.log.logger import logger @@ -117,14 +117,14 @@ class DirectoryReader: if not await file_path_.is_file(): return None try: - async with async_open(str(file_path_), encoding="utf-8") as file: + async with aiofiles.open(str(file_path_), encoding="utf-8") as file: # UnicodeDecodeError: 'charmap' codec can't decode byte 0x9d in position 3069: # character maps to return await file.read() except UnicodeDecodeError: # This is happening in Windows, so we need to open the file in binary mode # The file is always just a python file, so we can safely read it as utf-8 - async with async_open(str(file_path_), "rb") as f: + async with aiofiles.open(str(file_path_), "rb") as f: return (await f.read()).decode("utf-8") def get_files(self): diff --git a/src/lfx/src/lfx/load/load.py b/src/lfx/src/lfx/load/load.py index 17fe1b1b9b..7eb7ba17df 100644 --- a/src/lfx/src/lfx/load/load.py +++ b/src/lfx/src/lfx/load/load.py @@ -3,7 +3,7 @@ from io import StringIO from pathlib import Path from typing import TYPE_CHECKING -from aiofile import async_open +import aiofiles from dotenv import dotenv_values from lfx.graph.schema import RunOutputs @@ -57,7 +57,7 @@ async def aload_flow_from_json( # override env variables with .env file if env_file and tweaks is not None: - async with async_open(Path(env_file), encoding="utf-8") as f: + async with aiofiles.open(Path(env_file), encoding="utf-8") as f: content = await f.read() env_vars = dotenv_values(stream=StringIO(content)) tweaks = replace_tweaks_with_env(tweaks=tweaks, env_vars=env_vars) @@ -66,7 +66,7 @@ async def aload_flow_from_json( await update_settings(cache=cache) if isinstance(flow, str | Path): - async with async_open(Path(flow), encoding="utf-8") as f: + async with aiofiles.open(Path(flow), encoding="utf-8") as f: content = await f.read() flow_graph = json.loads(content) # If input is a dictionary, assume it's a JSON object diff --git a/src/lfx/src/lfx/services/settings/base.py b/src/lfx/src/lfx/services/settings/base.py index 08e9eabda5..d885c1ffa1 100644 --- a/src/lfx/src/lfx/services/settings/base.py +++ b/src/lfx/src/lfx/services/settings/base.py @@ -6,9 +6,9 @@ from pathlib import Path from shutil import copy2 from typing import Any, Literal +import aiofiles import orjson import yaml -from aiofile import async_open from pydantic import Field, field_validator from pydantic.fields import FieldInfo from pydantic_settings import BaseSettings, EnvSettingsSource, PydanticBaseSettingsSource, SettingsConfigDict @@ -705,7 +705,7 @@ async def load_settings_from_yaml(file_path: str) -> Settings: else: file_path_ = Path(file_path) - async with async_open(file_path_.name, encoding="utf-8") as f: + async with aiofiles.open(file_path_.name, encoding="utf-8") as f: content = await f.read() settings_dict = yaml.safe_load(content) settings_dict = {k.upper(): v for k, v in settings_dict.items()}