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 <manav2000@users.noreply.github.com>

* 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 <manav2000@users.noreply.github.com>

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

---------

Co-authored-by: manav2000 <manav2000@users.noreply.github.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: 3 people

src/backend/base/langflow/api/v1/flows_helpers.py

@@ -13,7 +13,7 @@
 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

src/backend/base/langflow/initial_setup/setup.py

@@ -13,11 +13,11 @@
 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)
 

src/backend/base/langflow/services/flow/flow_runner.py

@@ -3,7 +3,7 @@
 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 @@ async def database_exists_check():
     @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

src/backend/base/langflow/services/storage/local.py

@@ -5,7 +5,7 @@
 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 @@ async def save_file(self, flow_id: str, file_name: str, data: bytes, *, append:
 
         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 @@ async def get_file(self, flow_id: str, file_name: str) -> bytes:
             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 @@ async def get_file_stream(self, flow_id: str, file_name: str, chunk_size: int =
             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:

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)

src/backend/tests/unit/services/storage/test_local_storage_service.py

@@ -361,3 +361,28 @@ async def test_concurrent_file_operations(self, local_storage_service):
         # 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

src/lfx/src/lfx/custom/directory_reader/directory_reader.py

@@ -3,8 +3,8 @@
 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 @@ async def aread_file_content(self, file_path):
         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 <undefined>
                 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):

src/lfx/src/lfx/load/load.py

@@ -3,7 +3,7 @@
 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

src/lfx/src/lfx/services/settings/base.py

@@ -6,9 +6,9 @@
 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()}