Canvas Node Editor Backend: DAG-Based Workflow Executor

Canvas Node Editor Backend Implementation — Topological Sort–Based Workflow Executor

This post shares my experience implementing the backend for the canvas node editor, a core feature of the XGen platform. To execute the node graphs built with React Flow on the frontend, the backend needed a DAG (Directed Acyclic Graph)–based execution engine.

Workflow Execution Flow

When a user connects nodes on the frontend and clicks the run button, the backend goes through the following steps:

1. The frontend sends node/edge data as JSON
2. The backend constructs the DAG and validates it
3. Topological sort determines the execution order
4. Nodes are executed in order (or in parallel)
5. Results are streamed back via SSE

Implementing the Topological Sort Executor

from collections import deque, defaultdict
from typing import Dict, List, Set

class WorkflowExecutor:
    def __init__(self, nodes: List[dict], edges: List[dict]):
        self.nodes = {n["id"]: n for n in nodes}
        self.edges = edges
        self.graph: Dict[str, List[str]] = defaultdict(list)
        self.in_degree: Dict[str, int] = defaultdict(int)
        self._build_graph()

    def _build_graph(self):
        for node_id in self.nodes:
            self.in_degree.setdefault(node_id, 0)
        for edge in self.edges:
            src, tgt = edge["source"], edge["target"]
            self.graph[src].append(tgt)
            self.in_degree[tgt] += 1

    def _detect_cycle(self) -> bool:
        visited = set()
        rec_stack = set()

        def dfs(node: str) -> bool:
            visited.add(node)
            rec_stack.add(node)
            for neighbor in self.graph[node]:
                if neighbor not in visited:
                    if dfs(neighbor):
                        return True
                elif neighbor in rec_stack:
                    return True
            rec_stack.discard(node)
            return False

        for node_id in self.nodes:
            if node_id not in visited:
                if dfs(node_id):
                    return True
        return False

    def get_execution_order(self) -> List[List[str]]:
        if self._detect_cycle():
            raise ValueError("워크플로우에 순환 의존성이 있습니다")

        in_deg = dict(self.in_degree)
        queue = deque([n for n, d in in_deg.items() if d == 0])
        levels: List[List[str]] = []

        while queue:
            level = list(queue)
            levels.append(level)
            next_queue = deque()
            for node_id in level:
                for neighbor in self.graph[node_id]:
                    in_deg[neighbor] -= 1
                    if in_deg[neighbor] == 0:
                        next_queue.append(neighbor)
            queue = next_queue

        return levels

Passing Results Between Nodes

Once a node executes, its output needs to be forwarded to the next node. ExecutionContext handles this.

class ExecutionContext:
    def __init__(self):
        self.results: Dict[str, Any] = {}

    def set_result(self, node_id: str, output: Any):
        self.results[node_id] = output

    def get_input(self, node_id: str, edges: List[dict]) -> Dict:
        inputs = {}
        for edge in edges:
            if edge["target"] == node_id:
                src_handle = edge.get("sourceHandle", "output")
                tgt_handle = edge.get("targetHandle", "input")
                inputs[tgt_handle] = self.results.get(
                    edge["source"], {}
                ).get(src_handle)
        return inputs

Problems Encountered During Development

Cycle detection was the trickiest part. Initially I ran a simple BFS, but there were cases where cycles in complex graphs went undetected. Switching to DFS-based cycle detection resolved the issue and made the system reliable. For parallel execution, running all nodes at the same level concurrently with asyncio.gather yielded a significant performance improvement.