Fixes #206

state-manager/app/models/node_template_model.py

@@ -2,9 +2,14 @@
 from typing import Any, Optional, List
 
 
+class Unites(BaseModel):
+    identifier: str = Field(..., description="Identifier of the node")
+
+
 class NodeTemplate(BaseModel):
     node_name: str = Field(..., description="Name of the node")
     namespace: str = Field(..., description="Namespace of the node")
     identifier: str = Field(..., description="Identifier of the node")
     inputs: dict[str, Any] = Field(..., description="Inputs of the node")
-    next_nodes: Optional[List[str]] = Field(None, description="Next nodes to execute")
+    next_nodes: Optional[List[str]] = Field(None, description="Next nodes to execute")
+    unites: Optional[List[Unites]] = Field(None, description="Unites of the node")

state-manager/app/tasks/create_next_state.py

@@ -6,9 +6,13 @@
 from app.models.graph_template_validation_status import GraphTemplateValidationStatus
 from app.models.db.registered_node import RegisteredNode
 from app.models.state_status_enum import StateStatusEnum
+from beanie.operators import NE
+from app.singletons.logs_manager import LogsManager
 
 from json_schema_to_pydantic import create_model
 
+logger = LogsManager().get_logger()
+
 async def create_next_state(state: State):
     graph_template = None
 
@@ -50,6 +54,32 @@ async def create_next_state(state: State):
             if not next_node_template:
                 continue
 
+            depends_satisfied = True
+            if next_node_template.unites is not None and len(next_node_template.unites) > 0:
+                pending_count = 0
+                for depend in next_node_template.unites:
+                    if depend.identifier == state.identifier:
+                        continue
+                    else:
+                        root_parent = state.parents.get(depend.identifier)
+                        if root_parent is None:
+                            raise Exception(f"Root parent of {depend.identifier} not found")
+
+                        pending_count = await State.find(
+                            State.identifier == depend.identifier,
+                            State.namespace_name == state.namespace_name,
+                            State.graph_name == state.graph_name,
+                            NE(State.status, StateStatusEnum.SUCCESS),
+                            {f"parents.{depend.identifier}": parents[depend.identifier]}
+                        ).count()
+                    if pending_count > 0:
+                        logger.info(f"Node {next_node_template.identifier} depends on {depend.identifier} but it is not satisfied")
+                        depends_satisfied = False
+                        break
+
+            if not depends_satisfied:
+                continue
+
             registered_node = await RegisteredNode.find_one(RegisteredNode.name == next_node_template.node_name, RegisteredNode.namespace == next_node_template.namespace)
 
             if not registered_node:

state-manager/app/tasks/verify_graph.py

@@ -219,6 +219,18 @@ def dfs_visit(current_node: str, parent_node: str | None = None, current_path: l
 
     return dependency_graph
 
+async def verify_unites(graph_nodes: list[NodeTemplate], dependency_graph: dict | None, errors: list[str]):
+    if dependency_graph is None:
+        return
+
+    for node in graph_nodes:
+        if node.unites is None or len(node.unites) == 0:
+            continue
+        for depend in node.unites:
+            if depend.identifier not in dependency_graph[node.identifier]:
+                errors.append(f"Node {node.identifier} depends on {depend.identifier} which is not a dependency of {node.identifier}")
+
+
 async def verify_graph(graph_template: GraphTemplate):
     try:
         errors = []
@@ -229,16 +241,19 @@ async def verify_graph(graph_template: GraphTemplate):
         await verify_node_exists(graph_template.nodes, database_nodes, errors)
         await verify_node_identifiers(graph_template.nodes, errors)
         await verify_secrets(graph_template, database_nodes, errors)
-        dependency_graph = await verify_topology(graph_template.nodes, errors)   
+        dependency_graph = await verify_topology(graph_template.nodes, errors)
 
         if dependency_graph is not None and len(errors) == 0:        
             await verify_inputs(graph_template.nodes, database_nodes, dependency_graph, errors)
 
+        await verify_unites(graph_template.nodes, dependency_graph, errors)
+
         if errors or dependency_graph is None:
             graph_template.validation_status = GraphTemplateValidationStatus.INVALID
             graph_template.validation_errors = errors
             await graph_template.save()
             return
+
         graph_template.validation_status = GraphTemplateValidationStatus.VALID
         graph_template.validation_errors = None
         await graph_template.save()