Struct DistributedDagExecutor 

pub struct DistributedDagExecutor<Q: WorkQueuePort> { /* private fields */ }

Executes a DAG wave using a WorkQueuePort to distribute node-level tasks across workers.

Workers are spawned as Tokio tasks that pull from the queue, call the appropriate service, and acknowledge results. For local development the LocalWorkQueue is used; in production any queue backend can be plugged in without changing this executor.

Example

use stygian_graph::adapters::distributed::{DistributedDagExecutor, LocalWorkQueue};
use stygian_graph::ports::work_queue::WorkTask;

use stygian_graph::adapters::noop::NoopService;
use serde_json::json;
use std::sync::Arc;
use std::collections::HashMap;

let queue = Arc::new(LocalWorkQueue::new());
let executor = DistributedDagExecutor::new(queue, 4);

let mut services: HashMap<String, Arc<dyn stygian_graph::ports::ScrapingService>> =
    HashMap::new();
services.insert("noop".to_string(), Arc::new(NoopService));

let tasks = vec![WorkTask {
    id: "p1::fetch::01".to_string(),
    pipeline_id: "p1".to_string(),
    node_name: "fetch".to_string(),
    input: json!({"url": "https://example.com"}),
    wave: 0,
    attempt: 0,
    idempotency_key: "ik-01".to_string(),
}];

let results = executor.execute_wave("p1", tasks, &services).await.unwrap();
assert!(!results.is_empty() || results.is_empty()); // noop returns empty

Implementations

impl<Q: WorkQueuePort + 'static> DistributedDagExecutor<Q>

pub fn new(queue: Arc<Q>, worker_concurrency: usize) -> Self

Create a new executor with the given work queue and worker concurrency.

worker_concurrency controls how many parallel worker tasks drain the queue.

pub async fn execute_wave( &self, pipeline_id: &str, tasks: Vec<WorkTask>, services: &HashMap<String, Arc<dyn ScrapingService>>, ) -> Result<Vec<(String, Value)>>

Execute a single wave of tasks, distributing them across workers.

Returns (node_name, output) pairs for all tasks in the wave.