Struct DagExecutor 

pub struct DagExecutor { /* private fields */ }

DAG executor that processes pipeline graphs

Executes scraping pipelines as directed acyclic graphs using petgraph. Independent branches are executed concurrently using tokio::spawn. Data from upstream nodes is passed as input to downstream nodes.

Implementations

impl DagExecutor

pub fn new() -> Self

Create a new DAG executor

Example

use stygian_graph::domain::graph::DagExecutor;

let executor = DagExecutor::new();

pub fn from_pipeline(pipeline: &Pipeline) -> Result<Self, StygianError>

Build a graph from a pipeline definition

Errors

Returns GraphError::CycleDetected if the pipeline contains a cycle. Returns GraphError::NodeNotFound if an edge references an unknown node.

pub async fn execute( &self, services: &HashMap<String, Arc<dyn ScrapingService>>, ) -> Result<Vec<NodeResult>, StygianError>

Execute the pipeline using the provided service registry.

Nodes are executed in topological order. Independent nodes at the same depth are spawned concurrently via tokio::spawn. The output of each node is available to all downstream nodes as their ServiceInput.params.

Errors

Returns GraphError::ExecutionFailed if any node execution fails.

pub fn node_count(&self) -> usize

Get the total number of nodes in the graph

Example

use stygian_graph::domain::graph::{Pipeline, Node, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
assert_eq!(executor.node_count(), 2);

pub fn edge_count(&self) -> usize

Get the total number of edges in the graph

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));
pipeline.add_edge(Edge::new("a", "b"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
assert_eq!(executor.edge_count(), 1);

pub fn node_ids(&self) -> Vec<String>

Get all node IDs in the graph

Example

use stygian_graph::domain::graph::{Pipeline, Node, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("fetch", "http", json!({})));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
assert!(executor.node_ids().contains(&"fetch".to_string()));

pub fn get_node(&self, id: &str) -> Option<&Node>

Get a node by ID

Example

use stygian_graph::domain::graph::{Pipeline, Node, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("fetch", "http", json!({})));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let node = executor.get_node("fetch");
assert!(node.is_some());
assert_eq!(node.unwrap().service, "http");

pub fn predecessors(&self, id: &str) -> Vec<String>

Get the predecessors (upstream nodes) of a node

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));
pipeline.add_edge(Edge::new("a", "b"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let preds = executor.predecessors("b");
assert_eq!(preds, vec!["a".to_string()]);

pub fn successors(&self, id: &str) -> Vec<String>

Get the successors (downstream nodes) of a node

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));
pipeline.add_edge(Edge::new("a", "b"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let succs = executor.successors("a");
assert_eq!(succs, vec!["b".to_string()]);

pub fn topological_order(&self) -> Vec<String>

Get the topological order of nodes

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));
pipeline.add_edge(Edge::new("a", "b"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let order = executor.topological_order();
// "a" must appear before "b"
let a_pos = order.iter().position(|x| x == "a").unwrap();
let b_pos = order.iter().position(|x| x == "b").unwrap();
assert!(a_pos < b_pos);

pub fn execution_waves(&self) -> Vec<ExecutionWave>

Get execution waves (groups of nodes that can run concurrently)

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));
pipeline.add_node(Node::new("c", "http", json!({})));
pipeline.add_edge(Edge::new("a", "c"));
pipeline.add_edge(Edge::new("b", "c"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let waves = executor.execution_waves();
// Wave 0 contains "a" and "b" (can run concurrently)
// Wave 1 contains "c" (depends on both)
assert_eq!(waves.len(), 2);

pub fn node_info(&self, id: &str) -> Option<NodeInfo>

Get information about a specific node

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("fetch", "http", json!({"url": "https://example.com"})));
pipeline.add_node(Node::new("extract", "ai", json!({})));
pipeline.add_edge(Edge::new("fetch", "extract"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let info = executor.node_info("fetch").unwrap();
assert_eq!(info.service, "http");
assert_eq!(info.in_degree, 0);
assert_eq!(info.out_degree, 1);

pub fn query_nodes(&self, query: &NodeQuery) -> Vec<NodeInfo>

Get all nodes matching a query

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use stygian_graph::domain::introspection::NodeQuery;
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("fetch1", "http", json!({})));
pipeline.add_node(Node::new("fetch2", "http", json!({})));
pipeline.add_node(Node::new("extract", "ai", json!({})));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let http_nodes = executor.query_nodes(&NodeQuery::by_service("http"));
assert_eq!(http_nodes.len(), 2);

pub fn connectivity(&self) -> ConnectivityMetrics

Get connectivity metrics for the graph

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));
pipeline.add_edge(Edge::new("a", "b"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let metrics = executor.connectivity();
assert_eq!(metrics.root_nodes, vec!["a".to_string()]);
assert_eq!(metrics.leaf_nodes, vec!["b".to_string()]);

pub fn critical_path(&self) -> CriticalPath

Get the critical path (longest path through the graph)

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));
pipeline.add_node(Node::new("c", "http", json!({})));
pipeline.add_edge(Edge::new("a", "b"));
pipeline.add_edge(Edge::new("b", "c"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let critical = executor.critical_path();
assert_eq!(critical.length, 3);
assert_eq!(critical.nodes, vec!["a", "b", "c"]);

pub fn impact_analysis(&self, id: &str) -> ImpactAnalysis

Analyze the impact of changing a node

Returns all nodes that would be affected (upstream dependencies and downstream dependents, transitively).

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("a", "http", json!({})));
pipeline.add_node(Node::new("b", "http", json!({})));
pipeline.add_node(Node::new("c", "http", json!({})));
pipeline.add_edge(Edge::new("a", "b"));
pipeline.add_edge(Edge::new("b", "c"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let impact = executor.impact_analysis("b");
assert_eq!(impact.upstream, vec!["a".to_string()]);
assert_eq!(impact.downstream, vec!["c".to_string()]);

pub fn snapshot(&self) -> GraphSnapshot

Get a complete snapshot of the graph for introspection

Example

use stygian_graph::domain::graph::{Pipeline, Node, Edge, DagExecutor};
use serde_json::json;

let mut pipeline = Pipeline::new("test");
pipeline.add_node(Node::new("fetch", "http", json!({})));
pipeline.add_node(Node::new("extract", "ai", json!({})));
pipeline.add_edge(Edge::new("fetch", "extract"));

let executor = DagExecutor::from_pipeline(&pipeline).unwrap();
let snapshot = executor.snapshot();

assert_eq!(snapshot.node_count, 2);
assert_eq!(snapshot.edge_count, 1);

Trait Implementations

impl Default for DagExecutor

fn default() -> Self

Returns the “default value” for a type.