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.

Panics

Panics if an internal Mutex is poisoned (i.e. another thread panicked while holding the lock). Treat this as unrecoverable.

Errors

Returns StygianError when a service reports a failure, the executor is shut down, or a worker task cannot be enqueued.