From c721fcafc8b14292cdf90aaa653d03ea2074ecde Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Mon, 15 Jun 2026 16:04:01 -0700 Subject: [PATCH] fix: guard event serialization against unserializable component payloads (#12591) (#13651) * fix: guard event serialization against unserializable component payloads (#12591) Loop flows containing a vector store (e.g. Chroma) failed to build with: Error building Component Loop: [TypeError("'_thread.lock' object is not iterable"), TypeError('vars() argument must have __dict__ attribute')] The Loop runs its body in an isolated subgraph and streams an on_end_vertex event for every completed inner vertex. EventManager.send_event serialized that payload with a raw fastapi.jsonable_encoder call. When a payload carries an object jsonable_encoder cannot serialize -- most notably a threading.Lock held by a vector-DB client -- its last-resort fallback raises ValueError([TypeError, TypeError]), which propagated out of the subgraph and aborted the whole flow build. The earlier fix (#12877) only guarded ResultData.validate_model and build_output_logs (both already routed through the fail-safe serialize()), so this jsonable_encoder path was never covered and the issue persisted. send_event now falls back to lfx's fail-safe serialize() when jsonable_encoder raises, degrading unknown objects to a string representation instead of crashing. The happy path is unchanged (the fallback only runs on failure). Adds a focused EventManager unit test and an end-to-end Loop regression test whose body emits a lock-bearing Data (both fail without the fix). * [autofix.ci] apply automated fixes --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- src/lfx/src/lfx/events/event_manager.py | 14 +- .../test_loop_unserializable_output.py | 121 ++++++++++++++++++ .../tests/unit/events/test_event_manager.py | 32 +++++ 3 files changed, 166 insertions(+), 1 deletion(-) create mode 100644 src/lfx/tests/unit/components/flow_controls/test_loop_unserializable_output.py diff --git a/src/lfx/src/lfx/events/event_manager.py b/src/lfx/src/lfx/events/event_manager.py index 0b8004dea9..6649542ff4 100644 --- a/src/lfx/src/lfx/events/event_manager.py +++ b/src/lfx/src/lfx/events/event_manager.py @@ -12,6 +12,7 @@ from fastapi.encoders import jsonable_encoder from typing_extensions import Protocol from lfx.log.logger import logger +from lfx.serialization.serialization import serialize if TYPE_CHECKING: # Lightweight type stub for log types @@ -69,7 +70,18 @@ class EventManager: self.events[name] = callback_ def send_event(self, *, event_type: str, data: LoggableType): - jsonable_data = jsonable_encoder(data) + try: + jsonable_data = jsonable_encoder(data) + except (ValueError, TypeError) as exc: + # Component payloads can contain objects that FastAPI's jsonable_encoder + # cannot serialize — e.g. a vector-DB client holding a ``threading.Lock`` + # surfaced through a Loop's per-iteration build events (issue #12591). + # jsonable_encoder's last-resort fallback raises ``ValueError([TypeError, ...])`` + # there, which would otherwise abort the entire flow build. Degrade to the + # fail-safe serializer, which converts unknown objects to a string + # representation instead of crashing. + logger.debug(f"jsonable_encoder failed for event '{event_type}'; using safe serializer: {exc}") + jsonable_data = serialize(data, to_str=True) json_data = {"event": event_type, "data": jsonable_data} event_id = f"{event_type}-{uuid.uuid4()}" str_data = json.dumps(json_data) + "\n\n" diff --git a/src/lfx/tests/unit/components/flow_controls/test_loop_unserializable_output.py b/src/lfx/tests/unit/components/flow_controls/test_loop_unserializable_output.py new file mode 100644 index 0000000000..461a17332f --- /dev/null +++ b/src/lfx/tests/unit/components/flow_controls/test_loop_unserializable_output.py @@ -0,0 +1,121 @@ +"""Regression test for issue #12591: Loop + vector DB fails to build. + +A vector-store component inside a Loop body (e.g. Chroma) can surface objects +that FastAPI's ``jsonable_encoder`` cannot serialize — most notably a +``threading.Lock`` held by the underlying client. The Loop runs its body in an +isolated subgraph and streams an ``on_end_vertex`` event for every completed +inner vertex; that event payload was serialized with a raw ``jsonable_encoder`` +call, whose last-resort fallback raises +``ValueError([TypeError("'_thread.lock' object is not iterable"), +TypeError('vars() argument must have __dict__ attribute')])``. + +The exception propagated out of the subgraph and surfaced to the user as +``Error building Component Loop: [TypeError(...), TypeError(...)]``, breaking the +whole flow. This test drives a real Loop graph whose body emits a lock-bearing +``Data`` and asserts the build completes instead of crashing. +""" + +import asyncio +import json +import threading + +import pytest +from lfx.components.flow_controls.loop import LoopComponent +from lfx.custom.custom_component.component import Component +from lfx.events.event_manager import create_default_event_manager +from lfx.graph import Graph +from lfx.io import HandleInput, Output +from lfx.schema.data import Data +from lfx.schema.dataframe import DataFrame + + +class _LockEmittingComponent(Component): + """Loop-body node that emits a ``Data`` whose payload carries a lock. + + Stands in for a vector-DB component (Chroma/Astra/…) whose output or + metadata references a client object holding a non-serializable + ``threading.Lock``. + """ + + display_name = "Lock Emitter" + name = "LockEmitter" + + inputs = [HandleInput(name="input_value", display_name="In", input_types=["Data", "Message"])] + outputs = [Output(display_name="Out", name="out", method="build_out")] + + def build_out(self) -> Data: + result = Data(data={"text": "result", "vector_store": threading.Lock()}) + # Components surface their last output as ``status``; this is what feeds + # the build artifacts/events that get serialized for the UI. + self.status = result + return result + + +def _attach_loop_feedback(loop: LoopComponent, source: Component) -> None: + """Wire ``source.out`` back into ``loop.item`` in the frontend edge shape. + + Mirrors the helper used by the other loop graph tests: the loop feedback + edge has to be appended in the frontend-compatible shape and the source + component registered so ``Graph()`` walks to it. + """ + loop._edges.append( + { + "source": source.get_id(), + "target": loop.get_id(), + "data": { + "sourceHandle": { + "dataType": source.name, + "id": source.get_id(), + "name": "out", + "output_types": ["Data"], + }, + "targetHandle": { + "dataType": "LoopComponent", + "id": loop.get_id(), + "name": "item", + "output_types": ["Data", "Message"], + }, + }, + } + ) + if source not in loop._components: + loop._components.append(source) + + +@pytest.mark.asyncio +async def test_loop_with_unserializable_body_output_builds(): + """Loop body that emits a lock-bearing Data must not crash the build (#12591).""" + loop = LoopComponent(_id="loop") + rows = [Data(text=f"Row {i}") for i in range(3)] + loop.set(data=DataFrame(rows)) + + body = _LockEmittingComponent(_id="LockEmitter-body") + body.set(input_value=loop.item_output) + _attach_loop_feedback(loop, body) + + # A downstream consumer on `done` so done_output runs (classic Loop topology). + done_sink = LoopComponent(_id="loop-done-sink") + done_sink.set(data=loop.done_output) + + graph = Graph(loop, done_sink) + queue: asyncio.Queue = asyncio.Queue() + event_manager = create_default_event_manager(queue=queue) + + # Pre-fix this raised ComponentBuildError("Error building Component Loop: " + # "[TypeError(\"'_thread.lock' object is not iterable\"), ...]"). + results = [result async for result in graph.async_start(event_manager=event_manager)] + + loop_result = next(r for r in results if getattr(r, "vertex", None) and r.vertex.id == "loop") + assert loop_result.valid, "Loop vertex should build successfully despite unserializable body output" + + # The inner body vertex's end event was emitted (not dropped) once per row, + # with the unserializable lock degraded to a string instead of crashing. + end_vertex_events = [] + while not queue.empty(): + _, raw, _ = await queue.get() + payload = json.loads(raw.decode("utf-8").strip()) + if payload["event"] == "end_vertex": + end_vertex_events.append(payload["data"]) + + body_runs = [e for e in end_vertex_events if e.get("build_data", {}).get("id") == "LockEmitter-body"] + assert len(body_runs) == len(rows), f"Expected {len(rows)} body iterations, got {len(body_runs)}" diff --git a/src/lfx/tests/unit/events/test_event_manager.py b/src/lfx/tests/unit/events/test_event_manager.py index e8e0ecc66b..a38d5bd1fe 100644 --- a/src/lfx/tests/unit/events/test_event_manager.py +++ b/src/lfx/tests/unit/events/test_event_manager.py @@ -2,6 +2,7 @@ import asyncio import json +import threading from unittest.mock import MagicMock import pytest @@ -130,6 +131,37 @@ class TestEventManager: # Should not raise exception, just log debug message manager.send_event(event_type="test", data=test_data) + def test_send_event_degrades_unserializable_payload(self): + """Regression for #12591: an unserializable payload must not crash the build. + + A ``threading.Lock`` held by a vector-DB client (surfaced through a Loop's + per-iteration build events) makes FastAPI's jsonable_encoder raise + ``ValueError([TypeError, TypeError])``. ``send_event`` should fall back to the + fail-safe serializer and still emit a JSON-decodable event with the object + degraded to a string. + """ + queue = MagicMock() + manager = EventManager(queue) + + lock = threading.Lock() + # Mirror the on_end_vertex build_data shape emitted during a loop subgraph build. + data = {"build_data": {"id": "Chroma-1", "data": {"text": "chunk", "vector_store": lock}}} + + # Must not raise (pre-fix this propagated as "Error building Component Loop"). + manager.send_event(event_type="end_vertex", data=data) + + queue.put_nowait.assert_called_once() + _, data_bytes, _ = queue.put_nowait.call_args[0][0] + parsed = json.loads(data_bytes.decode("utf-8").strip()) + + assert parsed["event"] == "end_vertex" + # The lock survives as a string representation rather than crashing serialization. + serialized_lock = parsed["data"]["build_data"]["data"]["vector_store"] + assert isinstance(serialized_lock, str) + assert "lock" in serialized_lock.lower() + # Serializable siblings are preserved. + assert parsed["data"]["build_data"]["data"]["text"] == "chunk" + def test_noop_method(self): """Test noop method.""" queue = asyncio.Queue()