Merge branch 'worktree-agent-aa79b4c585b494d98' into fix/flow-debugger-arch-review
This commit is contained in:
@@ -16,14 +16,17 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
||||
/// takes the first row's identity, time, input and title, the latest row's outcome, subtitle, output and error;
|
||||
/// <c>DurationMs</c> is the sync row's own, or submit→terminal wall time for a cached operation (null while pending);
|
||||
/// <c>Lifecycle</c> lists every row in time order.</item>
|
||||
/// <item>Steps are ordered by <c>OccurredAtUtc</c>, then a kind rank (inbound before route call
|
||||
/// before script run before writes/waits before outbound), then <c>EventId</c>.</item>
|
||||
/// <item><b>Order</b> = emission order between executions, timestamp within one (arch-review F5: central and
|
||||
/// site clocks differ). Executions are ordered by the input index of their earliest-given row; inside one
|
||||
/// execution — one clock — steps are ordered by <c>OccurredAtUtc</c>, then a kind rank (inbound before route
|
||||
/// call before script run before writes/waits before outbound), then the input index. Replay input arrives
|
||||
/// timestamp-sorted from the repository, so its order is unchanged; the Live path's collector order wins.</item>
|
||||
/// <item>Within one <c>ExecutionId</c> consecutive steps are chained in that order, <b>skipping async steps</b> — a
|
||||
/// <c>Notify</c>, or a cached operation whose first row is <c>CachedSubmit</c>. Each skipped step instead gets an
|
||||
/// <c>Async</c> (dashed) edge from the chain step immediately before it in time.</item>
|
||||
/// <item>The first step of an execution is linked from its spawner: a <c>RouteCall</c> whose
|
||||
/// <c>childExecutionId</c> matches, else the latest step of the parent execution that occurred
|
||||
/// no later than it — that second case is a <c>Spawn</c> edge (trigger cascade). The edge
|
||||
/// <c>childExecutionId</c> matches, else the last step of the parent execution that precedes it in
|
||||
/// the order above — that second case is a <c>Spawn</c> edge (trigger cascade). The edge
|
||||
/// <c>CrossesBoundary</c> when exactly one end is in the Central lane.</item>
|
||||
/// <item>An <c>AttributeWait</c> that was satisfied gets an <c>External</c> (dashed) edge from the
|
||||
/// preceding <c>AttributeWrite</c> in the same execution — the device echo the system inferred.</item>
|
||||
@@ -41,12 +44,47 @@ public static class FlowGraphBuilder
|
||||
if (events.Count == 0)
|
||||
return new FlowGraph(Array.Empty<FlowStep>(), Array.Empty<FlowEdge>(), Predicted: false, Truncated: truncated);
|
||||
|
||||
var steps = Fold(events)
|
||||
.OrderBy(s => s.OccurredAtUtc)
|
||||
// Arch-review F5: a step's position in the input is the only cross-node ordering we can trust —
|
||||
// central rows carry central's clock, site rows the site's, and a 1-2 s skew used to sort a whole
|
||||
// site execution ahead of the inbound row that started it.
|
||||
var seq = new Dictionary<Guid, int>(events.Count);
|
||||
for (var i = 0; i < events.Count; i++)
|
||||
seq.TryAdd(events[i].EventId, i);
|
||||
|
||||
// A folded step keeps the identity of its earliest-by-TIME row, which need not be the row given
|
||||
// first — so a folded node's index is the minimum over the rows folded into it.
|
||||
int InputIndex(FlowStep step)
|
||||
{
|
||||
var index = seq.TryGetValue(step.EventId, out var own) ? own : int.MaxValue;
|
||||
if (step.Lifecycle is { Count: > 1 } rows)
|
||||
{
|
||||
foreach (var row in rows)
|
||||
{
|
||||
if (seq.TryGetValue(row.EventId, out var i) && i < index)
|
||||
index = i;
|
||||
}
|
||||
}
|
||||
|
||||
return index;
|
||||
}
|
||||
|
||||
var folded = Fold(events).ToList();
|
||||
var firstSeqByExecution = folded
|
||||
.GroupBy(s => s.ExecutionId ?? Guid.Empty)
|
||||
.ToDictionary(g => g.Key, g => g.Min(InputIndex));
|
||||
var steps = folded
|
||||
.OrderBy(s => firstSeqByExecution[s.ExecutionId ?? Guid.Empty]) // executions in emission order
|
||||
.ThenBy(s => s.OccurredAtUtc) // one clock inside an execution
|
||||
.ThenBy(s => KindRank(s.Kind))
|
||||
.ThenBy(s => s.EventId)
|
||||
.ThenBy(InputIndex)
|
||||
.ToList();
|
||||
|
||||
// Position in the ordered list — the "before" relation the spawner search uses, so a skewed
|
||||
// clock can no longer push every candidate parent step "after" the child.
|
||||
var position = new Dictionary<Guid, int>(steps.Count);
|
||||
for (var i = 0; i < steps.Count; i++)
|
||||
position[steps[i].EventId] = i;
|
||||
|
||||
var edges = new List<FlowEdge>();
|
||||
var byExecution = steps
|
||||
.Where(s => s.ExecutionId is not null)
|
||||
@@ -93,8 +131,12 @@ public static class FlowGraphBuilder
|
||||
if (routeCallsByChild.TryGetValue(executionId, out var routeCall))
|
||||
spawner = routeCall;
|
||||
else if (byExecution.TryGetValue(parentId, out var parentChain))
|
||||
spawner = parentChain.LastOrDefault(p => p.OccurredAtUtc <= first.OccurredAtUtc && p.Kind != FlowStepKind.Response && !IsAsync(p))
|
||||
{
|
||||
// The parent step that precedes the child in DISPLAY order, not by raw timestamp (F5).
|
||||
var childPosition = position[first.EventId];
|
||||
spawner = parentChain.LastOrDefault(p => position[p.EventId] < childPosition && p.Kind != FlowStepKind.Response && !IsAsync(p))
|
||||
?? parentChain[0];
|
||||
}
|
||||
|
||||
if (spawner is null) continue;
|
||||
var spawn = spawner.Kind != FlowStepKind.RouteCall;
|
||||
|
||||
@@ -173,8 +173,12 @@ public sealed class FlowDebugRunner
|
||||
result.ErrorMessage ?? "Access denied: routed call outside your site scope.");
|
||||
}
|
||||
|
||||
// Arch-review F5: the collector's emission order is authoritative — central rows carry central's
|
||||
// clock and site rows the site's, so re-sorting by OccurredAtUtc lets a skewed site clock float
|
||||
// its steps ahead of the RouteCall that started them. (The value-change path below keeps its
|
||||
// OrderBy: every row it merges was stamped by the same site.)
|
||||
var ordered = new List<AuditEvent>(events.Count + 1) { inboundRow };
|
||||
ordered.AddRange(events.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId));
|
||||
ordered.AddRange(events);
|
||||
|
||||
return new FlowDebugRunResult(
|
||||
root, result.Success, result.ErrorMessage, result.Success ? null : result.ErrorCode,
|
||||
|
||||
@@ -46,8 +46,10 @@ public class FlowGraphBuilderTests
|
||||
executionId: Cascade, parentExecutionId: SiteRun, sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 10,
|
||||
extra: new ScriptTraceExtra { TriggerKind = "valuechange", TriggerAttribute = "DelmiaReceiver.Trigger", TriggerValue = "false" }.ToJson());
|
||||
|
||||
// Deliberately shuffled: the builder must order by time.
|
||||
return [wait, cascade, inbound, write2, run, route, write1];
|
||||
// Deliberately shuffled WITHIN each execution: the builder must order by time inside one execution.
|
||||
// The executions themselves are given in emission order (root, site run, cascade) — since arch-review
|
||||
// F5 that is what fixes their relative order, because they do not share a clock.
|
||||
return [inbound, route, wait, run, write2, write1, cascade];
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -365,6 +367,105 @@ public class FlowGraphBuilderTests
|
||||
Assert.Equal(graph.Steps[1].EventId, edge.ToEventId);
|
||||
}
|
||||
|
||||
// ---- Arch-review F5: emission order between executions, timestamps within one ------------
|
||||
|
||||
/// <summary>
|
||||
/// Central and site clocks differ. The four rows are given in EMISSION order (the Live path's
|
||||
/// collector order); the graph must keep the executions in that order whichever way the skew runs.
|
||||
/// </summary>
|
||||
private static List<AuditEvent> SkewedRouteShape(int siteSkewMs)
|
||||
{
|
||||
var inbound = ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiInbound, AuditKind.InboundRequest, AuditStatus.Delivered,
|
||||
occurredAtUtc: T0, actor: "k", target: "M", sourceNode: "central-a", executionId: Root, durationMs: 900,
|
||||
requestSummary: "{}", responseSummary: "{\"ok\":true}");
|
||||
var route = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.RouteCall, AuditStatus.Delivered,
|
||||
occurredAtUtc: T0.AddMilliseconds(50), target: "M1.S", sourceNode: "central-a", executionId: Root,
|
||||
extra: new ScriptTraceExtra { ChildExecutionId = SiteRun }.ToJson());
|
||||
var run = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
||||
occurredAtUtc: T0.AddMilliseconds(siteSkewMs), actor: "S", target: "S", sourceNode: "node-a",
|
||||
executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 20,
|
||||
extra: new ScriptTraceExtra { TriggerKind = "call" }.ToJson());
|
||||
var write = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.AttributeWrite, AuditStatus.Delivered,
|
||||
occurredAtUtc: T0.AddMilliseconds(siteSkewMs + 10), actor: "S", target: "A", sourceNode: "node-a",
|
||||
executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 2,
|
||||
extra: new ScriptTraceExtra { Writes = [new AttributeWriteDetail("A", "1", true, null)] }.ToJson());
|
||||
return [inbound, route, run, write];
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SiteClockAhead_SiteStepsStillFollowTheirRouteCall_AndPrecedeResponse()
|
||||
{
|
||||
var events = SkewedRouteShape(siteSkewMs: 2000); // site 2 s ahead of central
|
||||
var graph = FlowGraphBuilder.Build(events);
|
||||
|
||||
Assert.Equal(
|
||||
new[]
|
||||
{
|
||||
FlowStepKind.InboundRequest, FlowStepKind.RouteCall, FlowStepKind.ScriptRun,
|
||||
FlowStepKind.AttributeWrite, FlowStepKind.Response,
|
||||
},
|
||||
graph.Steps.Select(s => s.Kind));
|
||||
Assert.Equal(events.Select(e => e.EventId), graph.Steps.Take(4).Select(s => s.EventId));
|
||||
// The Response time still comes off the inbound row's own clock (inbound + DurationMs).
|
||||
Assert.Equal(T0.AddMilliseconds(900), graph.Steps[4].OccurredAtUtc.UtcDateTime);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SiteClockBehind_SiteStepsDoNotSortAheadOfTheInboundRequest()
|
||||
{
|
||||
var events = SkewedRouteShape(siteSkewMs: -2000); // site 2 s behind central
|
||||
var graph = FlowGraphBuilder.Build(events);
|
||||
|
||||
Assert.Equal(
|
||||
new[]
|
||||
{
|
||||
FlowStepKind.InboundRequest, FlowStepKind.RouteCall, FlowStepKind.ScriptRun,
|
||||
FlowStepKind.AttributeWrite, FlowStepKind.Response,
|
||||
},
|
||||
graph.Steps.Select(s => s.Kind));
|
||||
Assert.Equal(events.Select(e => e.EventId), graph.Steps.Take(4).Select(s => s.EventId));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SiteClockBehind_TriggerCascade_SpawnsFromTheLastParentStep_NotTheFirst()
|
||||
{
|
||||
// Mixed-version shape: no RouteCall row, so the spawner is found by position, and a site
|
||||
// clock behind central used to push every parent step "after" the child → parentChain[0].
|
||||
var inbound = ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiInbound, AuditKind.InboundRequest, AuditStatus.Delivered,
|
||||
occurredAtUtc: T0, actor: "k", target: "M", sourceNode: "central-a", executionId: Root, durationMs: 900);
|
||||
var centralDb = ScadaBridgeAuditEventFactory.Create(AuditChannel.DbOutbound, AuditKind.DbWrite, AuditStatus.Delivered,
|
||||
occurredAtUtc: T0.AddMilliseconds(10), actor: "api:M", target: "CfgDb", sourceNode: "central-a", executionId: Root);
|
||||
var run = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
||||
occurredAtUtc: T0.AddMilliseconds(-2000), actor: "S", target: "S", sourceNode: "node-a",
|
||||
executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 20,
|
||||
extra: new ScriptTraceExtra { TriggerKind = "call" }.ToJson());
|
||||
|
||||
var graph = FlowGraphBuilder.Build([inbound, centralDb, run]);
|
||||
|
||||
var spawn = Assert.Single(graph.Edges, e => e.Spawn);
|
||||
Assert.Equal(centralDb.EventId, spawn.FromEventId); // the LAST parent step, not the inbound row
|
||||
Assert.Equal(run.EventId, spawn.ToEventId);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void FoldedNode_TakesTheEmissionIndexOfItsEarliestEmittedRow()
|
||||
{
|
||||
// The folded step keeps the earliest-by-TIME row's id (the resolve here is emitted first),
|
||||
// so its sequence must be the minimum input index over the rows folded into it.
|
||||
var corr = Guid.NewGuid();
|
||||
var resolve = Outbound(AuditChannel.ApiOutbound, AuditKind.CachedResolve, AuditStatus.Delivered, 4000, corr, "Delmia.Ack", response: "res");
|
||||
var submit = Outbound(AuditChannel.ApiOutbound, AuditKind.CachedSubmit, AuditStatus.Submitted, 35, corr, "Delmia.Ack", request: "req");
|
||||
var run = SiteRunRow(20, Cascade);
|
||||
|
||||
// Emission order interleaves the executions: the folded node's earliest row is index 0, so
|
||||
// its execution leads — even though the id the folded step keeps (the submit) is index 2.
|
||||
var graph = FlowGraphBuilder.Build([resolve, run, submit]);
|
||||
|
||||
var api = Assert.Single(graph.Steps, s => s.Kind == FlowStepKind.ApiCall);
|
||||
Assert.Equal(submit.EventId, api.EventId);
|
||||
Assert.Equal(new[] { FlowStepKind.ApiCall, FlowStepKind.ScriptRun }, graph.Steps.Select(s => s.Kind));
|
||||
}
|
||||
|
||||
// ---- Review findings on Task 2 (2026-09-06) --------------------------------------------
|
||||
|
||||
[Fact]
|
||||
|
||||
@@ -124,6 +124,31 @@ public class FlowDebugRunnerTests
|
||||
Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.RouteCall);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RunMethod_SiteClockBehindCentral_StillReturnsTheSinkEventsInEmissionOrder()
|
||||
{
|
||||
// Arch-review F5: the site's rows carry the SITE clock. A site 2 s behind central used to
|
||||
// re-sort ahead of the always-on central RouteCall row.
|
||||
var method = new ApiMethod("M", "x") { Id = 6, TimeoutSeconds = 5 };
|
||||
_apiRepo.GetApiMethodByIdAsync(6, Arg.Any<CancellationToken>()).Returns(method);
|
||||
_locator.GetSiteIdForInstanceAsync("inst", Arg.Any<CancellationToken>()).Returns("site-a");
|
||||
var child = Guid.NewGuid();
|
||||
var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
||||
occurredAtUtc: DateTime.UtcNow.AddSeconds(-2), executionId: child, target: "work");
|
||||
_router.RouteToCallAsync("site-a", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
|
||||
.Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 7, null, DateTimeOffset.UtcNow, child, [siteRun]));
|
||||
_executor.RegisterHandler("M", ctx => ctx.Route.To("inst").Call("work"));
|
||||
|
||||
var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
|
||||
new FlowDebugRunMethodCommand(6, null), Deployer, CancellationToken.None);
|
||||
|
||||
Assert.True(result.Completed, result.Error);
|
||||
Assert.Equal(3, result.Events.Count);
|
||||
Assert.Equal(AuditKind.InboundRequest, result.Events[0].AsRow().Kind);
|
||||
Assert.Equal(AuditKind.RouteCall, result.Events[1].AsRow().Kind);
|
||||
Assert.Same(siteRun, result.Events[2]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RunMethod_UnknownMethod_IsNotFound()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user