feat(engine): Executor fan-out + failure propagation + status rollup
This commit is contained in:
Claude
@@ -96,12 +96,14 @@ public sealed class Executor
|
||||
return eid;
|
||||
}
|
||||
|
||||
private enum ArrivalKind { Data, Failure }
|
||||
private sealed record Arrival(ArrivalKind Kind, IReadOnlyList<Item> Items);
|
||||
|
||||
private async Task RunAsync(
|
||||
string executionId, string workflowId, WorkflowGraph graph,
|
||||
string triggerNodeId, IReadOnlyList<Item> triggerItems,
|
||||
Guid? answerChannelId, ILogger logger, CancellationToken ct)
|
||||
{
|
||||
// 1. Record trigger node_execution (inputs=[], outputs=[triggerItems]).
|
||||
var triggerNode = graph.Nodes.FirstOrDefault(n => n.Id == triggerNodeId)
|
||||
?? throw new InvalidOperationException($"Trigger node {triggerNodeId} not found in graph.");
|
||||
var triggerStart = DateTimeOffset.UtcNow;
|
||||
@@ -113,74 +115,146 @@ public sealed class Executor
|
||||
endedAt: DateTimeOffset.UtcNow,
|
||||
error: null);
|
||||
|
||||
// Arrivals table: (toNode, toPort) -> items
|
||||
var arrivals = new Dictionary<(string node, int port), IReadOnlyList<Item>>();
|
||||
var arrivalCounts = new Dictionary<string, int>(StringComparer.Ordinal);
|
||||
var arrivals = new Dictionary<string, Dictionary<int, Arrival>>(StringComparer.Ordinal);
|
||||
foreach (var n in graph.Nodes) arrivals[n.Id] = new();
|
||||
var inDegree = graph.Nodes.ToDictionary(n => n.Id,
|
||||
n => graph.Edges.Count(e => e.ToNode == n.Id), StringComparer.Ordinal);
|
||||
|
||||
void Deliver(string toNode, int toPort, IReadOnlyList<Item> items)
|
||||
var ready = new Queue<string>();
|
||||
var running = new Dictionary<Task<NodeResult>, string>();
|
||||
var completedCount = 0;
|
||||
var failedCount = 0;
|
||||
var skippedCount = 0;
|
||||
|
||||
void Deliver(string toNode, int toPort, Arrival a)
|
||||
{
|
||||
arrivals[(toNode, toPort)] = items;
|
||||
arrivalCounts[toNode] = arrivalCounts.GetValueOrDefault(toNode) + 1;
|
||||
arrivals[toNode][toPort] = a;
|
||||
if (arrivals[toNode].Count == inDegree[toNode]) ready.Enqueue(toNode);
|
||||
}
|
||||
|
||||
foreach (var e in graph.Edges.Where(e => e.FromNode == triggerNodeId && e.FromPort == 0))
|
||||
Deliver(e.ToNode, e.ToPort, triggerItems);
|
||||
Deliver(e.ToNode, e.ToPort, new Arrival(ArrivalKind.Data, triggerItems));
|
||||
|
||||
// 2. Compute per-node in-degree = number of distinct incoming ports.
|
||||
var inDegree = graph.Nodes.ToDictionary(n => n.Id,
|
||||
n => graph.Edges.Count(e => e.ToNode == n.Id),
|
||||
StringComparer.Ordinal);
|
||||
|
||||
// 3. Scheduler loop.
|
||||
var ready = new Queue<string>();
|
||||
foreach (var (nid, need) in arrivalCounts)
|
||||
if (arrivals.Keys.Count(k => k.node == nid) == inDegree[nid])
|
||||
ready.Enqueue(nid);
|
||||
|
||||
var successes = 0;
|
||||
while (ready.Count > 0)
|
||||
while (ready.Count > 0 || running.Count > 0)
|
||||
{
|
||||
var nid = ready.Dequeue();
|
||||
var node = graph.Nodes.First(n => n.Id == nid);
|
||||
var typ = _nodes.Get(node.TypeId);
|
||||
if (typ is not IActionNode action)
|
||||
throw new InvalidOperationException($"Node {nid} is not an action type.");
|
||||
|
||||
// Build inputsByPort — for v1 arrival counts always equal InputCount when ready.
|
||||
var byPort = new IReadOnlyList<Item>[Math.Max(1, action.InputCount)];
|
||||
for (var p = 0; p < byPort.Length; p++)
|
||||
byPort[p] = arrivals.TryGetValue((nid, p), out var items) ? items : Array.Empty<Item>();
|
||||
|
||||
using var cfgDoc = JsonDocument.Parse(node.ConfigJson);
|
||||
var ctx = new ExecutionContext(
|
||||
executionId, workflowId, nid, cfgDoc.RootElement.Clone(),
|
||||
answerChannelId, _lf.CreateLogger($"node.{nid}"),
|
||||
node.ConnectionId, _instances);
|
||||
|
||||
var startedAt = DateTimeOffset.UtcNow;
|
||||
var outputs = await action.ExecuteAsync(ctx, byPort, ct);
|
||||
var endedAt = DateTimeOffset.UtcNow;
|
||||
|
||||
_neRepo.Create(executionId, nid,
|
||||
inputsJson: SerializeInputs(byPort),
|
||||
outputsJson: SerializeOutputs(outputs),
|
||||
status: "completed",
|
||||
startedAt: startedAt,
|
||||
endedAt: endedAt,
|
||||
error: null);
|
||||
successes++;
|
||||
|
||||
// Deliver on output edges.
|
||||
foreach (var e in graph.Edges.Where(e => e.FromNode == nid))
|
||||
while (ready.Count > 0)
|
||||
{
|
||||
if (e.FromPort >= outputs.Count) continue;
|
||||
Deliver(e.ToNode, e.ToPort, outputs[e.FromPort]);
|
||||
if (arrivals.Keys.Count(k => k.node == e.ToNode) == inDegree[e.ToNode])
|
||||
ready.Enqueue(e.ToNode);
|
||||
var nid = ready.Dequeue();
|
||||
var node = graph.Nodes.First(n => n.Id == nid);
|
||||
var portMap = arrivals[nid];
|
||||
|
||||
// Skip if ANY incoming was a failure.
|
||||
if (portMap.Values.Any(v => v.Kind == ArrivalKind.Failure))
|
||||
{
|
||||
var startedAt = DateTimeOffset.UtcNow;
|
||||
_neRepo.Create(executionId, nid,
|
||||
inputsJson: SerializeInputs(BuildInputsByPort(portMap, allowFailurePlaceholders: true)),
|
||||
outputsJson: "[]",
|
||||
status: "skipped_due_to_failure",
|
||||
startedAt: startedAt,
|
||||
endedAt: startedAt,
|
||||
error: null);
|
||||
skippedCount++;
|
||||
foreach (var e in graph.Edges.Where(e => e.FromNode == nid))
|
||||
Deliver(e.ToNode, e.ToPort, new Arrival(ArrivalKind.Failure, Array.Empty<Item>()));
|
||||
continue;
|
||||
}
|
||||
|
||||
var byPort = BuildInputsByPort(portMap, allowFailurePlaceholders: false);
|
||||
var typ = _nodes.Get(node.TypeId);
|
||||
if (typ is not IActionNode action)
|
||||
throw new InvalidOperationException($"Node {nid} is not an action type.");
|
||||
|
||||
running[RunNodeAsync(executionId, workflowId, node, action, byPort, answerChannelId, ct)] = nid;
|
||||
}
|
||||
|
||||
if (running.Count == 0) break;
|
||||
var finished = await Task.WhenAny(running.Keys);
|
||||
var nidDone = running[finished];
|
||||
running.Remove(finished);
|
||||
var result = await finished;
|
||||
|
||||
_neRepo.Create(executionId, nidDone,
|
||||
inputsJson: result.InputsJson,
|
||||
outputsJson: result.Success ? SerializeOutputs(result.Outputs!) : "[]",
|
||||
status: result.Success ? "completed" : "failed",
|
||||
startedAt: result.StartedAt,
|
||||
endedAt: result.EndedAt,
|
||||
error: result.Error);
|
||||
|
||||
if (result.Success)
|
||||
{
|
||||
completedCount++;
|
||||
foreach (var e in graph.Edges.Where(e => e.FromNode == nidDone))
|
||||
{
|
||||
if (e.FromPort >= result.Outputs!.Count) continue;
|
||||
Deliver(e.ToNode, e.ToPort, new Arrival(ArrivalKind.Data, result.Outputs[e.FromPort]));
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
failedCount++;
|
||||
foreach (var e in graph.Edges.Where(e => e.FromNode == nidDone))
|
||||
Deliver(e.ToNode, e.ToPort, new Arrival(ArrivalKind.Failure, Array.Empty<Item>()));
|
||||
}
|
||||
}
|
||||
|
||||
_exRepo.SetStatus(executionId, "completed", null, DateTimeOffset.UtcNow);
|
||||
// Status roll-up (spec §4):
|
||||
// completed = no failed nodes
|
||||
// failed = at least one failed + zero non-trigger completions
|
||||
// partial = otherwise (failures + successes both present)
|
||||
var status = failedCount == 0
|
||||
? "completed"
|
||||
: (completedCount == 0 ? "failed" : "partial");
|
||||
_exRepo.SetStatus(executionId, status, error: null, endedAt: DateTimeOffset.UtcNow);
|
||||
}
|
||||
|
||||
private static IReadOnlyList<IReadOnlyList<Item>> BuildInputsByPort(
|
||||
Dictionary<int, Arrival> portMap, bool allowFailurePlaceholders)
|
||||
{
|
||||
var count = portMap.Count == 0 ? 1 : portMap.Keys.Max() + 1;
|
||||
var byPort = new IReadOnlyList<Item>[count];
|
||||
for (var p = 0; p < count; p++)
|
||||
{
|
||||
if (portMap.TryGetValue(p, out var a))
|
||||
byPort[p] = a.Items;
|
||||
else if (allowFailurePlaceholders)
|
||||
byPort[p] = Array.Empty<Item>();
|
||||
else
|
||||
byPort[p] = Array.Empty<Item>();
|
||||
}
|
||||
return byPort;
|
||||
}
|
||||
|
||||
private sealed record NodeResult(
|
||||
bool Success,
|
||||
IReadOnlyList<IReadOnlyList<Item>>? Outputs,
|
||||
string InputsJson,
|
||||
DateTimeOffset StartedAt,
|
||||
DateTimeOffset EndedAt,
|
||||
string? Error);
|
||||
|
||||
private async Task<NodeResult> RunNodeAsync(
|
||||
string executionId, string workflowId, WorkflowNode node, IActionNode action,
|
||||
IReadOnlyList<IReadOnlyList<Item>> byPort, Guid? answerChannelId, CancellationToken ct)
|
||||
{
|
||||
var inputsJson = SerializeInputs(byPort);
|
||||
using var cfgDoc = JsonDocument.Parse(node.ConfigJson);
|
||||
var ctx = new ExecutionContext(
|
||||
executionId, workflowId, node.Id, cfgDoc.RootElement.Clone(),
|
||||
answerChannelId, _lf.CreateLogger($"node.{node.Id}"),
|
||||
node.ConnectionId, _instances);
|
||||
|
||||
var startedAt = DateTimeOffset.UtcNow;
|
||||
try
|
||||
{
|
||||
var outputs = await action.ExecuteAsync(ctx, byPort, ct);
|
||||
return new NodeResult(true, outputs, inputsJson, startedAt, DateTimeOffset.UtcNow, null);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return new NodeResult(false, null, inputsJson, startedAt, DateTimeOffset.UtcNow, ex.Message);
|
||||
}
|
||||
}
|
||||
|
||||
private static string SerializeInputs(IReadOnlyList<IReadOnlyList<Item>> ports)
|
||||
|
||||
Reference in New Issue
Block a user