feat(engine): Executor — linear happy-path (trigger → action → terminal)
This commit is contained in:
Claude
194
src/VolcanicCore.Api/Engine/Executor.cs
Normal file
194
src/VolcanicCore.Api/Engine/Executor.cs
Normal file
@@ -0,0 +1,194 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using VolcanicCore.Api.Registries;
|
||||
using VolcanicCore.Core;
|
||||
using VolcanicCore.Persistence.Repositories;
|
||||
|
||||
namespace VolcanicCore.Api.Engine;
|
||||
|
||||
public sealed record WorkflowNode(string Id, string TypeId, string? ConnectionId, string ConfigJson);
|
||||
public sealed record WorkflowEdge(string FromNode, int FromPort, string ToNode, int ToPort);
|
||||
public sealed record WorkflowGraph
|
||||
{
|
||||
public IReadOnlyList<WorkflowNode> Nodes { get; init; } = Array.Empty<WorkflowNode>();
|
||||
public IReadOnlyList<WorkflowEdge> Edges { get; init; } = Array.Empty<WorkflowEdge>();
|
||||
|
||||
public static WorkflowGraph Parse(string graphJson)
|
||||
{
|
||||
var opts = new JsonSerializerOptions(JsonSerializerDefaults.Web)
|
||||
{
|
||||
PropertyNameCaseInsensitive = true,
|
||||
};
|
||||
using var doc = JsonDocument.Parse(graphJson);
|
||||
var root = doc.RootElement;
|
||||
var nodes = new List<WorkflowNode>();
|
||||
foreach (var n in root.GetProperty("nodes").EnumerateArray())
|
||||
{
|
||||
var id = n.GetProperty("id").GetString()!;
|
||||
var typeId = n.TryGetProperty("type_id", out var t1) ? t1.GetString()
|
||||
: n.TryGetProperty("typeId", out var t2) ? t2.GetString() : null;
|
||||
var connId = n.TryGetProperty("connection_id", out var c1) ? c1.GetString()
|
||||
: n.TryGetProperty("connectionId", out var c2) ? c2.GetString() : null;
|
||||
var cfg = n.TryGetProperty("config_json", out var g1) ? g1.GetRawText()
|
||||
: n.TryGetProperty("config", out var g2) ? g2.GetRawText() : "{}";
|
||||
nodes.Add(new WorkflowNode(id, typeId!, connId, cfg));
|
||||
}
|
||||
var edges = new List<WorkflowEdge>();
|
||||
if (root.TryGetProperty("edges", out var edgesEl))
|
||||
{
|
||||
foreach (var e in edgesEl.EnumerateArray())
|
||||
{
|
||||
var from = e.GetProperty("from_node").GetString()!;
|
||||
var to = e.GetProperty("to_node").GetString()!;
|
||||
var fp = e.TryGetProperty("from_port", out var fpe) ? fpe.GetInt32() : 0;
|
||||
var tp = e.TryGetProperty("to_port", out var tpe) ? tpe.GetInt32() : 0;
|
||||
edges.Add(new WorkflowEdge(from, fp, to, tp));
|
||||
}
|
||||
}
|
||||
return new WorkflowGraph { Nodes = nodes, Edges = edges };
|
||||
}
|
||||
}
|
||||
|
||||
public sealed class Executor
|
||||
{
|
||||
private readonly ExecutionRepository _exRepo;
|
||||
private readonly NodeExecutionRepository _neRepo;
|
||||
private readonly NodeRegistry _nodes;
|
||||
private readonly IConnectionInstanceRegistry _instances;
|
||||
private readonly ILoggerFactory _lf;
|
||||
|
||||
public Executor(
|
||||
ExecutionRepository exRepo,
|
||||
NodeExecutionRepository neRepo,
|
||||
NodeRegistry nodes,
|
||||
IConnectionInstanceRegistry instances,
|
||||
ILoggerFactory lf)
|
||||
{
|
||||
_exRepo = exRepo;
|
||||
_neRepo = neRepo;
|
||||
_nodes = nodes;
|
||||
_instances = instances;
|
||||
_lf = lf;
|
||||
}
|
||||
|
||||
public async Task<string> StartExecutionAsync(
|
||||
string workflowId,
|
||||
WorkflowGraph graph,
|
||||
string triggerNodeId,
|
||||
string triggerKind,
|
||||
IReadOnlyList<Item> triggerItems,
|
||||
JsonElement? triggerData,
|
||||
Guid? answerChannelId,
|
||||
CancellationToken ct)
|
||||
{
|
||||
var triggerDataJson = triggerData?.GetRawText() ?? "null";
|
||||
var eid = _exRepo.Create(workflowId, triggerKind, triggerDataJson);
|
||||
var logger = _lf.CreateLogger($"exec.{eid}");
|
||||
|
||||
try
|
||||
{
|
||||
await RunAsync(eid, workflowId, graph, triggerNodeId, triggerItems, answerChannelId, logger, ct);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_exRepo.SetStatus(eid, "failed", ex.Message, DateTimeOffset.UtcNow);
|
||||
}
|
||||
return eid;
|
||||
}
|
||||
|
||||
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;
|
||||
_neRepo.Create(executionId, triggerNodeId,
|
||||
inputsJson: "[]",
|
||||
outputsJson: SerializeOutputs(new[] { triggerItems }),
|
||||
status: "completed",
|
||||
startedAt: triggerStart,
|
||||
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);
|
||||
|
||||
void Deliver(string toNode, int toPort, IReadOnlyList<Item> items)
|
||||
{
|
||||
arrivals[(toNode, toPort)] = items;
|
||||
arrivalCounts[toNode] = arrivalCounts.GetValueOrDefault(toNode) + 1;
|
||||
}
|
||||
|
||||
foreach (var e in graph.Edges.Where(e => e.FromNode == triggerNodeId && e.FromPort == 0))
|
||||
Deliver(e.ToNode, e.ToPort, 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)
|
||||
{
|
||||
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))
|
||||
{
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
_exRepo.SetStatus(executionId, "completed", null, DateTimeOffset.UtcNow);
|
||||
}
|
||||
|
||||
private static string SerializeInputs(IReadOnlyList<IReadOnlyList<Item>> ports)
|
||||
=> JsonSerializer.Serialize(ports.Select(port => port.Select(i => new
|
||||
{
|
||||
json = i.Json,
|
||||
binary = i.Binary?.ToDictionary(kv => kv.Key, kv => Convert.ToBase64String(kv.Value)),
|
||||
}).ToArray()).ToArray());
|
||||
|
||||
private static string SerializeOutputs(IReadOnlyList<IReadOnlyList<Item>> ports) => SerializeInputs(ports);
|
||||
}
|
||||
Reference in New Issue
Block a user