fix: Topological sort on child spans

Author: erichare

src/backend/base/langflow/services/tracing/native.py

@@ -320,25 +320,13 @@ async def _flush_to_database(self, error: Exception | None = None) -> None:
                 )
                 await session.merge(trace)
 
-                for span_data in self.completed_spans:
-                    try:
-                        span_uuid = UUID_(span_data["id"])
-                    except (ValueError, TypeError):
-                        # Span IDs from LangChain callbacks are strings, not UUIDs — derive
-                        # a stable UUID so the same span always maps to the same DB row.
-                        span_uuid = uuid5(LANGFLOW_SPAN_NAMESPACE, f"{self.trace_id}-{span_data['id']}")
-
-                    parent_uuid = None
-                    if span_data.get("parent_span_id"):
-                        parent_id = span_data["parent_span_id"]
-                        if isinstance(parent_id, UUID_):
-                            parent_uuid = parent_id
-                        else:
-                            try:
-                                parent_uuid = UUID_(str(parent_id))
-                            except (ValueError, TypeError):
-                                parent_uuid = uuid5(LANGFLOW_SPAN_NAMESPACE, f"{self.trace_id}-{parent_id}")
+                # Pre-compute UUIDs and topologically sort so parents are inserted
+                # before children — required by PostgreSQL's immediate FK enforcement
+                # on span.parent_span_id → span.id.
+                resolved = self._resolve_span_uuids(self.completed_spans)
+                resolved = self._topological_sort_spans(resolved)
 
+                for span_data, span_uuid, parent_uuid in resolved:
                     span = SpanTable(
                         id=span_uuid,
                         trace_id=self.trace_id,
@@ -553,6 +541,92 @@ def _build_completed_span(
             span["parent_span_id"] = parent_span_id
         return span
 
+    def _resolve_span_uuids(
+        self,
+        completed_spans: list[dict[str, Any]],
+    ) -> list[tuple[dict[str, Any], UUID, UUID | None]]:
+        """Pre-compute DB UUIDs for each span and its parent.
+
+        Returns a list of (span_data, span_uuid, parent_uuid) tuples that can
+        be topologically sorted before insertion.
+        """
+        from uuid import UUID as UUID_
+
+        resolved: list[tuple[dict[str, Any], UUID, UUID | None]] = []
+        for span_data in completed_spans:
+            try:
+                span_uuid = UUID_(span_data["id"])
+            except (ValueError, TypeError):
+                span_uuid = uuid5(LANGFLOW_SPAN_NAMESPACE, f"{self.trace_id}-{span_data['id']}")
+
+            parent_uuid: UUID | None = None
+            if span_data.get("parent_span_id"):
+                parent_id = span_data["parent_span_id"]
+                if isinstance(parent_id, UUID_):
+                    parent_uuid = parent_id
+                else:
+                    try:
+                        parent_uuid = UUID_(str(parent_id))
+                    except (ValueError, TypeError):
+                        parent_uuid = uuid5(LANGFLOW_SPAN_NAMESPACE, f"{self.trace_id}-{parent_id}")
+
+            resolved.append((span_data, span_uuid, parent_uuid))
+        return resolved
+
+    @staticmethod
+    def _topological_sort_spans(
+        resolved: list[tuple[dict[str, Any], UUID, UUID | None]],
+    ) -> list[tuple[dict[str, Any], UUID, UUID | None]]:
+        """Sort spans so parents appear before children.
+
+        PostgreSQL enforces foreign-key constraints at INSERT time, so a child
+        span referencing ``parent_span_id`` will fail if the parent row hasn't
+        been written yet.  This performs a Kahn's-algorithm-style topological
+        sort over the batch so that every parent is inserted first.
+
+        Spans whose ``parent_span_id`` points outside the current batch are
+        treated as roots (the parent already exists in the DB from a prior
+        flush).
+        """
+        batch_ids = {span_uuid for _, span_uuid, _ in resolved}
+
+        sorted_spans: list[tuple[dict[str, Any], UUID, UUID | None]] = []
+        inserted: set[UUID] = set()
+        remaining = list(resolved)
+
+        while remaining:
+            next_round: list[tuple[dict[str, Any], UUID, UUID | None]] = []
+            progress = False
+            for item in remaining:
+                _, span_uuid, parent_uuid = item
+                # Insert if: no parent, parent outside batch, or parent already inserted
+                if parent_uuid is None or parent_uuid not in batch_ids or parent_uuid in inserted:
+                    sorted_spans.append(item)
+                    inserted.add(span_uuid)
+                    progress = True
+                else:
+                    next_round.append(item)
+
+            if not progress:
+                # Cycle or unresolvable dependency detected.
+                # To avoid reintroducing foreign-key violations, break the cycle by
+                # nulling out parent_span_id for the remaining spans before inserting.
+                if next_round:
+                    logger.warning(
+                        "Detected cycle or unresolvable span dependencies in tracing batch; "
+                        "breaking parent relationships for %d spans to preserve DB integrity.",
+                        len(next_round),
+                    )
+                for span_data, span_uuid, _parent_uuid in next_round:
+                    # Ensure the payload reflects the broken parent relationship.
+                    if isinstance(span_data, dict):
+                        span_data["parent_span_id"] = None
+                    sorted_spans.append((span_data, span_uuid, None))
+                break
+            remaining = next_round
+
+        return sorted_spans
+
     @staticmethod
     def _map_trace_type(trace_type: str) -> SpanType:
         """Normalise Langflow's string trace types to the SpanType enum, defaulting to CHAIN for unknown values."""

src/backend/tests/unit/services/tracing/test_native_tracer.py

@@ -604,3 +604,171 @@ def test_callback_has_no_parent_span_id_when_no_component(self):
         assert callback is not None
         assert isinstance(callback, NativeCallbackHandler)
         assert callback.parent_span_id is None
+
+
+# ---------------------------------------------------------------------------
+# _topological_sort_spans
+# ---------------------------------------------------------------------------
+
+
+class TestTopologicalSortSpans:
+    """Verify that spans are sorted so parents appear before children.
+
+    This is critical for PostgreSQL which enforces FK constraints at INSERT time.
+    """
+
+    @staticmethod
+    def _make_span(span_id, parent_id=None) -> tuple[dict, UUID, UUID | None]:
+        """Helper to create a resolved (span_data, span_uuid, parent_uuid) tuple."""
+        span_data = {"id": str(span_id), "name": f"span-{span_id}"}
+        return (span_data, span_id, parent_id)
+
+    def test_no_parents(self):
+        a, b, c = uuid4(), uuid4(), uuid4()
+        items = [self._make_span(a), self._make_span(b), self._make_span(c)]
+        result = NativeTracer._topological_sort_spans(items)
+        assert [r[1] for r in result] == [a, b, c]
+
+    def test_child_after_parent(self):
+        parent_id = uuid4()
+        child_id = uuid4()
+        # Child comes first in the input list
+        items = [self._make_span(child_id, parent_id), self._make_span(parent_id)]
+        result = NativeTracer._topological_sort_spans(items)
+        uuids = [r[1] for r in result]
+        assert uuids.index(parent_id) < uuids.index(child_id)
+
+    def test_deep_nesting(self):
+        root = uuid4()
+        mid = uuid4()
+        leaf = uuid4()
+        # Reverse order: leaf, mid, root
+        items = [
+            self._make_span(leaf, mid),
+            self._make_span(mid, root),
+            self._make_span(root),
+        ]
+        result = NativeTracer._topological_sort_spans(items)
+        uuids = [r[1] for r in result]
+        assert uuids.index(root) < uuids.index(mid) < uuids.index(leaf)
+
+    def test_parent_outside_batch(self):
+        """Spans referencing a parent not in the batch are treated as roots."""
+        external_parent = uuid4()
+        child_id = uuid4()
+        items = [self._make_span(child_id, external_parent)]
+        result = NativeTracer._topological_sort_spans(items)
+        assert len(result) == 1
+        assert result[0][1] == child_id
+
+    def test_mixed_roots_and_children(self):
+        root_a = uuid4()
+        child_a = uuid4()
+        root_b = uuid4()
+        items = [
+            self._make_span(child_a, root_a),
+            self._make_span(root_b),
+            self._make_span(root_a),
+        ]
+        result = NativeTracer._topological_sort_spans(items)
+        uuids = [r[1] for r in result]
+        assert uuids.index(root_a) < uuids.index(child_a)
+
+    def test_cycle_two_node(self):
+        """Spans forming a 2-node cycle should not cause errors or drop spans."""
+        a = uuid4()
+        b = uuid4()
+        items = [
+            self._make_span(a, b),
+            self._make_span(b, a),
+        ]
+        result = NativeTracer._topological_sort_spans(items)
+        # Ensure all spans are present and no infinite loop / exception occurs.
+        uuids = [r[1] for r in result]
+        assert len(uuids) == 2
+        assert set(uuids) == {a, b}
+
+    def test_self_parent_span(self):
+        """A span that lists itself as its own parent should still be returned."""
+        span_id = uuid4()
+        items = [
+            self._make_span(span_id, span_id),
+        ]
+        result = NativeTracer._topological_sort_spans(items)
+        uuids = [r[1] for r in result]
+        assert len(uuids) == 1
+        assert uuids[0] == span_id
+
+    def test_empty_input(self):
+        result = NativeTracer._topological_sort_spans([])
+        assert result == []
+
+# ---------------------------------------------------------------------------
+# _flush_to_database with parent/child spans
+# ---------------------------------------------------------------------------
+
+
+class TestFlushParentChildOrder:
+    @pytest.mark.asyncio
+    async def test_flush_inserts_parent_before_child(self):
+        """Verify that the DB merge order respects parent -> child ordering."""
+        flow_id = str(uuid4())
+        tracer = _make_tracer(flow_id=flow_id)
+
+        parent_uuid = uuid4()
+        child_uuid = uuid4()
+
+        # Deliberately add child first to simulate the problematic ordering
+        tracer.completed_spans = [
+            {
+                "id": str(child_uuid),
+                "name": "Child Span",
+                "span_type": SpanType.CHAIN,
+                "inputs": {},
+                "outputs": None,
+                "start_time": datetime.now(tz=timezone.utc),
+                "end_time": datetime.now(tz=timezone.utc),
+                "latency_ms": 5,
+                "status": SpanStatus.OK,
+                "error": None,
+                "attributes": {},
+                "span_source": "component",
+                "parent_span_id": parent_uuid,
+            },
+            {
+                "id": str(parent_uuid),
+                "name": "Parent Span",
+                "span_type": SpanType.CHAIN,
+                "inputs": {},
+                "outputs": None,
+                "start_time": datetime.now(tz=timezone.utc),
+                "end_time": datetime.now(tz=timezone.utc),
+                "latency_ms": 10,
+                "status": SpanStatus.OK,
+                "error": None,
+                "attributes": {},
+                "span_source": "component",
+            },
+        ]
+
+        merged_objects = []
+        mock_session = AsyncMock()
+        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_session.__aexit__ = AsyncMock(return_value=False)
+
+        async def capture_merge(obj):
+            merged_objects.append(obj)
+
+        mock_session.merge = capture_merge
+
+        with patch("lfx.services.deps.session_scope", return_value=mock_session):
+            await tracer._flush_to_database()
+
+        from langflow.services.database.models.traces.model import SpanTable
+
+        span_objects = [o for o in merged_objects if isinstance(o, SpanTable)]
+        assert len(span_objects) == 2
+        # Parent (no parent_span_id) must come before child
+        assert span_objects[0].id == parent_uuid
+        assert span_objects[1].id == child_uuid
+        assert span_objects[1].parent_span_id == parent_uuid