DAG fan‑in node can execute more than once when dependencies finish close together

[next-n]

Summary
When a DAG node has multiple incoming dependencies, the fan‑in path checks “are all deps done?” and then executes the node. If two dependencies finish nearly at the same time, both can pass the check and both execute the fan‑in node, causing duplicate execution.

Location
service.rs — fan‑in branch (see TODO about the “done” check).

Steps to reproduce (conceptual)

Build a DAG where C depends on A and B.
Make A and B complete quickly.
Run the workflow; C can execute twice (log/side‑effect).
Expected
C runs exactly once after all dependencies complete.

Actual
C can be scheduled/run more than once under race conditions.

Note
I found this TODO in the code and would like to work on a fix + test if that’s okay.

[geofmureithi]

Your link points to a file url.
But I do know that that issue as that TODO was added in #668

Sure you can work on it, but I would feel better if you outlined the steps you want to take before submitting the code, especially if there are API changes.

Happy coding!

[next-n]

yes, I edit the GitHub URL of file location, sorry for confusion earlier. I will check again for the direction and reply here for confirmation first.

[next-n]

I plan to add a guard inside the fan-in all Done branch, before executor.call(req), to ensure only one predecessor executes the fan-in node.
That would be like the following.

// Multiple incoming nodes, fan-in scenario
_ => {
    let dependency_task_ids =
        context.get_dependency_task_ids(&incoming_nodes);

    let results = backend
        .check_status(
            dependency_task_ids.values().cloned().collect::<Vec<_>>(),
        )
        .await?;

    if results.iter().all(|s| matches!(s.status, Status::Done)) {

        // ===== FIX START =====
        let designated_parent = incoming_nodes
            .iter()
            .min_by_key(|n| n.index())
            .copied();

        let is_designated = context.prev_node == designated_parent;

        if !is_designated {
            return Ok(DagExecutionResponse::WaitingForDependencies {
                pending_dependencies: dependency_task_ids,
            });
        }
        // ===== FIX END =====

        let sorted_results = {
            // existing code unchanged
        };

        let encoded_input = B::Codec::encode(&res).map_err(|e| e.into())?;
        let req = req.map(|_| encoded_input);
        let response = executor.call(req).await?;
        (response, context)
    } else {
        return Ok(DagExecutionResponse::WaitingForDependencies {
            pending_dependencies: dependency_task_ids,
        });
    }
}

Can you give suggestions upon that fix?

[geofmureithi]

Interesting fix, I think it makes sense, please do a PR and include a test to check this. Let me know if you need help with the testing part.