dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(commons): FlowGraphBuilder review fixes — leading async steps link from the spawner, robust async rule, lifecycle tie-break, total DbSubtitle

Browse Source 
Review of Task 2 (2026-09-06 outbound-calls plan): (1) an outbound step that
precedes the first chain step of its execution — the default Replay shape when
a run's ScriptTrace rows are absent — was orphaned or given a solid spawner
edge; leading async steps now hang off the spawner on Async edges and the
first non-async step gets the normal link. (2) IsAsync keyed on the first row
being CachedSubmit; it now mirrors the projector's sync rule (first row not
ApiCall/DbWrite) so a group missing its submit row still branches. (3) Same-
tick rows tie-broke by GUID; new LifecycleRank (submit < attempt < resolve,
non-terminal < terminal) makes first/latest attribution order-independent.
(4) DbSubtitle used GetInt32 (FormatException on out-of-range/fractional) —
now TryGetInt32, never throws. Also: External-lane edges never count as
central↔site crossings; DurationMs clamped; FlowStep doc updated.

Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
This commit is contained in:
Joseph Doherty
2026-09-06 01:49:40 -04:00
parent 30faee0d0e
commit 865c4be651
5 changed files with 183 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Commons/Types/Debugging/FlowGraphBuilder.cs
+66 -15
View File
@@ -53,8 +53,12 @@ public static class FlowGraphBuilder
.GroupBy(s => s.ExecutionId!.Value)
.ToDictionary(g => g.Key, g => g.ToList());
// 1. Chain consecutive non-async steps inside each execution; async steps branch off the step before them.
foreach (var chain in byExecution.Values)
// 1. Chain consecutive non-async steps inside each execution; async steps branch off the step
// before them. Async steps that precede the first chain step ("leading" — the default Replay
// shape when a run has no ScriptTrace rows and only its outbound rows survive) are linked
// from the execution's spawner in step 2 instead, so nothing is orphaned.
var leadingAsync = new Dictionary<Guid, List<FlowStep>>();
foreach (var (executionId, chain) in byExecution)
{
FlowStep? prev = null;
foreach (var step in chain)
@@ -63,6 +67,8 @@ public static class FlowGraphBuilder
{
if (prev is not null)
edges.Add(new FlowEdge(prev.EventId, step.EventId, CrossesBoundary(prev, step), External: false, Async: true));
else
(leadingAsync.TryGetValue(executionId, out var l) ? l : leadingAsync[executionId] = []).Add(step);
continue;
}
@@ -90,9 +96,19 @@ public static class FlowGraphBuilder
spawner = parentChain.LastOrDefault(p => p.OccurredAtUtc <= first.OccurredAtUtc && p.Kind != FlowStepKind.Response && !IsAsync(p))
?? parentChain[0];
if (spawner is not null && spawner.EventId != first.EventId)
edges.Add(new FlowEdge(spawner.EventId, first.EventId, CrossesBoundary(spawner, first), External: false,
Spawn: spawner.Kind != FlowStepKind.RouteCall));
if (spawner is null) continue;
var spawn = spawner.Kind != FlowStepKind.RouteCall;
// Leading async steps hang off the spawner on Async edges; the first chain step (if any) gets the normal link.
if (leadingAsync.TryGetValue(executionId, out var leading))
{
foreach (var a in leading)
edges.Add(new FlowEdge(spawner.EventId, a.EventId, CrossesBoundary(spawner, a), External: false, Async: true, Spawn: spawn));
}
var firstChained = chain.FirstOrDefault(s => !IsAsync(s));
if (firstChained is not null && spawner.EventId != firstChained.EventId)
edges.Add(new FlowEdge(spawner.EventId, firstChained.EventId, CrossesBoundary(spawner, firstChained), External: false, Spawn: spawn));
}
// 3. Satisfied wait ⇐ preceding write in the same execution (inferred device echo).
@@ -148,13 +164,44 @@ public static class FlowGraphBuilder
}
/// <summary>
/// A step the script did not wait for: a notification, or a cached operation (first lifecycle row
/// <c>CachedSubmit</c>). Such steps hang off the chain on an <see cref="FlowEdge.Async"/> edge (D4).
/// A step the script did not wait for: a notification, or a cached operation — any API / DB step whose
/// first lifecycle row is not the sync one-shot kind (<c>ApiCall</c> / <c>DbWrite</c>), the same rule the
/// projector uses for the <c>sync</c> / <c>cached</c> subtitle. Such steps hang off the chain on an
/// <see cref="FlowEdge.Async"/> edge (D4).
/// </summary>
public static bool IsAsync(FlowStep step)
=> step.Kind == FlowStepKind.Notify
|| (step.Kind is FlowStepKind.ApiCall or FlowStepKind.DbCall
&& string.Equals(step.Lifecycle?.FirstOrDefault()?.Kind, nameof(AuditKind.CachedSubmit), StringComparison.Ordinal));
{
if (step.Kind == FlowStepKind.Notify) return true;
if (step.Kind is not (FlowStepKind.ApiCall or FlowStepKind.DbCall)) return false;
var firstKind = step.Lifecycle?.FirstOrDefault()?.Kind;
return firstKind is not null
&& !string.Equals(firstKind, nameof(AuditKind.ApiCall), StringComparison.Ordinal)
&& !string.Equals(firstKind, nameof(AuditKind.DbWrite), StringComparison.Ordinal);
}
/// <summary>
/// Orders the rows of one outbound operation when timestamps tie: submit before attempts before the
/// resolve, then non-terminal before terminal statuses — so "first" and "latest" (D3) never fall back
/// to GUID order for rows written in the same tick.
/// </summary>
public static int LifecycleRank(string? kind, string? status)
{
var k = kind switch
{
nameof(AuditKind.CachedSubmit) or nameof(AuditKind.NotifySend) => 0,
nameof(AuditKind.ApiCallCached) or nameof(AuditKind.DbWriteCached) or nameof(AuditKind.NotifyDeliver) => 1,
nameof(AuditKind.CachedResolve) => 2,
_ => 1,
};
var st = status switch
{
nameof(AuditStatus.Submitted) => 0,
nameof(AuditStatus.Forwarded) => 1,
nameof(AuditStatus.Attempted) => 2,
_ => 3,
};
return k * 4 + st;
}
/// <summary>
/// Projects every event and folds the outbound rows of one operation (same kind + <c>CorrelationId</c>) into one step
@@ -200,16 +247,18 @@ public static class FlowGraphBuilder
var key = ((FlowStepKind, Guid))item;
var rows = groups[key]
.OrderBy(r => r.Step.OccurredAtUtc)
.ThenBy(r => r.Step.EventId)
.Select(r => (r.Step, Row: FlowStepProjector.ToLifecycleRow(r.Evt)))
.OrderBy(r => r.Row.OccurredAtUtc)
.ThenBy(r => LifecycleRank(r.Row.Kind, r.Row.Status))
.ThenBy(r => r.Row.EventId)
.ToList();
var first = rows[0].Step;
var latest = rows[^1].Step;
var lifecycle = rows.Select(r => FlowStepProjector.ToLifecycleRow(r.Evt)).ToList();
var lifecycle = rows.Select(r => r.Row).ToList();
var terminal = latest.Outcome != FlowStepOutcome.Pending;
int? duration = rows.Count == 1
? first.DurationMs
: terminal ? (int?)Math.Max(0, (int)Math.Round((latest.OccurredAtUtc - first.OccurredAtUtc).TotalMilliseconds)) : null;
: terminal ? (int?)Math.Min(int.MaxValue, Math.Max(0, Math.Round((latest.OccurredAtUtc - first.OccurredAtUtc).TotalMilliseconds))) : null;
yield return first with
{
@@ -237,8 +286,10 @@ public static class FlowGraphBuilder
_ => 7,
};
/// <summary>Central ↔ site only; a hop onto the External lane is drawn by its own edge style, never as a cluster crossing.</summary>
private static bool CrossesBoundary(FlowStep a, FlowStep b)
=> (a.Lane == FlowLane.Central) != (b.Lane == FlowLane.Central);
=> a.Lane != FlowLane.External && b.Lane != FlowLane.External
&& (a.Lane == FlowLane.Central) != (b.Lane == FlowLane.Central);
/// <summary>A stable id for the synthesized Response node so re-builds of the same events select the same node.</summary>
private static Guid DeriveResponseId(Guid inboundEventId)
src/ZB.MOM.WW.ScadaBridge.Commons/Types/Debugging/FlowStep.cs
+4 -2
View File
@@ -76,8 +76,10 @@ public sealed record FlowLifecycleRow(
/// <summary>
/// One node of the flow graph — a central-side projection of a <c>ScriptTrace</c> /
/// <c>ApiInbound.InboundRequest</c> <see cref="ZB.MOM.WW.Audit.AuditEvent"/>. Never serialized
/// to the wire; the wire carries the audit events themselves (design §5.4).
/// <c>ApiInbound.InboundRequest</c> <see cref="ZB.MOM.WW.Audit.AuditEvent"/>, or of one outbound
/// operation's <c>ApiOutbound</c> / <c>DbOutbound</c> / <c>Notification</c> rows folded together
/// (<see cref="Lifecycle"/>; outbound-calls plan 2026-09-06 D3). Never serialized to the wire;
/// the wire carries the audit events themselves (design §5.4).
/// </summary>
public sealed record FlowStep(
Guid EventId,
src/ZB.MOM.WW.ScadaBridge.Commons/Types/Debugging/FlowStepProjector.cs
+2 -2
View File
@@ -164,8 +164,8 @@ public static class FlowStepProjector
{
var op = opEl.GetString();
int? rows = null;
if (root.TryGetProperty("rowsAffected", out var ra) && ra.ValueKind == JsonValueKind.Number) rows = ra.GetInt32();
else if (root.TryGetProperty("rowsReturned", out var rr) && rr.ValueKind == JsonValueKind.Number) rows = rr.GetInt32();
if (root.TryGetProperty("rowsAffected", out var ra) && ra.ValueKind == JsonValueKind.Number && ra.TryGetInt32(out var affected)) rows = affected;
else if (root.TryGetProperty("rowsReturned", out var rr) && rr.ValueKind == JsonValueKind.Number && rr.TryGetInt32(out var returned)) rows = returned;
if (rows is { } n && n >= 0)
return $"{op} · {n} {(n == 1 ? "row" : "rows")}";
return op;
tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Debugging/FlowGraphBuilderTests.cs
+98
View File
@@ -363,4 +363,102 @@ public class FlowGraphBuilderTests
Assert.Equal(graph.Steps[0].EventId, edge.FromEventId);
Assert.Equal(graph.Steps[1].EventId, edge.ToEventId);
}
// ---- Review findings on Task 2 (2026-09-06) --------------------------------------------
[Fact]
public void LeadingAsyncSteps_HangOffTheSpawner_NeverOrphaned()
{
// A routed run whose ScriptTrace rows are absent (TraceAllCalls off / older site): only its outbound rows survive.
var route = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.RouteCall, AuditStatus.Delivered,
occurredAtUtc: T0.AddMilliseconds(10), target: "M1.S", sourceNode: "central-a", executionId: Root,
extra: new ScriptTraceExtra { ChildExecutionId = SiteRun }.ToJson());
var cached = ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiOutbound, AuditKind.CachedSubmit, AuditStatus.Submitted,
occurredAtUtc: T0.AddMilliseconds(20), target: "Delmia.Ack", correlationId: Guid.NewGuid(), executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a");
var notify = ScadaBridgeAuditEventFactory.Create(AuditChannel.Notification, AuditKind.NotifySend, AuditStatus.Submitted,
occurredAtUtc: T0.AddMilliseconds(30), target: "Planners", correlationId: Guid.NewGuid(), executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a");
var sync = ScadaBridgeAuditEventFactory.Create(AuditChannel.DbOutbound, AuditKind.DbWrite, AuditStatus.Delivered,
occurredAtUtc: T0.AddMilliseconds(40), target: "MesDb", executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a");
var graph = FlowGraphBuilder.Build([route, cached, notify, sync]);
var routeStep = graph.Steps[0];
var reached = graph.Edges.Select(e => e.ToEventId).ToHashSet();
Assert.All(graph.Steps.Skip(1), s => Assert.Contains(s.EventId, reached));
Assert.Equal(2, graph.Edges.Count(e => e.Async && e.FromEventId == routeStep.EventId));
var syncEdge = Assert.Single(graph.Edges, e => e.ToEventId == graph.Steps[3].EventId);
Assert.Equal(routeStep.EventId, syncEdge.FromEventId);
Assert.False(syncEdge.Async);
Assert.False(syncEdge.Spawn);
Assert.Equal(3, graph.Edges.Count);
}
[Fact]
public void OnlyNotifyRows_InAnExecution_AllLinkFromTheSpawner()
{
var run = SiteRunRow(20);
var n1 = ScadaBridgeAuditEventFactory.Create(AuditChannel.Notification, AuditKind.NotifySend, AuditStatus.Submitted,
occurredAtUtc: T0.AddMilliseconds(30), target: "A", correlationId: Guid.NewGuid(), executionId: Cascade, parentExecutionId: SiteRun, sourceSiteId: "site-a");
var n2 = ScadaBridgeAuditEventFactory.Create(AuditChannel.Notification, AuditKind.NotifySend, AuditStatus.Submitted,
occurredAtUtc: T0.AddMilliseconds(31), target: "B", correlationId: Guid.NewGuid(), executionId: Cascade, parentExecutionId: SiteRun, sourceSiteId: "site-a");
var graph = FlowGraphBuilder.Build([run, n1, n2]);
Assert.Equal(2, graph.Edges.Count);
Assert.All(graph.Edges, e => { Assert.True(e.Async); Assert.True(e.Spawn); Assert.Equal(graph.Steps[0].EventId, e.FromEventId); });
}
[Fact]
public void Fold_IsOrderIndependent_AndTiesBreakByLifecycleRank()
{
var corr = Guid.NewGuid();
var submit = Outbound(AuditChannel.ApiOutbound, AuditKind.CachedSubmit, AuditStatus.Submitted, 35, corr, "Delmia.Ack", request: "req");
var resolve = Outbound(AuditChannel.ApiOutbound, AuditKind.CachedResolve, AuditStatus.Delivered, 35, corr, "Delmia.Ack", response: "res");
var a = FlowGraphBuilder.Build([SiteRunRow(20), resolve, submit]);
var b = FlowGraphBuilder.Build([SiteRunRow(20), submit, resolve]);
var sa = Assert.Single(a.Steps, s => s.Kind == FlowStepKind.ApiCall);
var sb = Assert.Single(b.Steps, s => s.Kind == FlowStepKind.ApiCall);
Assert.Equal(submit.EventId, sa.EventId);
Assert.Equal(sa.EventId, sb.EventId);
Assert.Equal("req", sa.InputJson);
Assert.Equal("res", sa.OutputJson);
Assert.Equal(FlowStepOutcome.Ok, sa.Outcome);
Assert.Equal(0, sa.DurationMs);
Assert.True(FlowGraphBuilder.IsAsync(sa));
Assert.Equal(sa.Lifecycle!.Select(r => r.Kind), sb.Lifecycle!.Select(r => r.Kind));
}
[Fact]
public void CachedStep_WhoseSubmitRowIsMissing_IsStillAsync()
{
var corr = Guid.NewGuid();
var graph = FlowGraphBuilder.Build([
SiteRunRow(20),
Outbound(AuditChannel.ApiOutbound, AuditKind.ApiCallCached, AuditStatus.Attempted, 3000, corr, "Delmia.Ack"),
Outbound(AuditChannel.ApiOutbound, AuditKind.CachedResolve, AuditStatus.Delivered, 4000, corr, "Delmia.Ack"),
]);
var api = Assert.Single(graph.Steps, s => s.Kind == FlowStepKind.ApiCall);
Assert.True(FlowGraphBuilder.IsAsync(api));
Assert.Single(graph.Edges, e => e.Async && e.ToEventId == api.EventId);
}
[Fact]
public void ExternalLaneEdges_NeverCountAsClusterBoundaryCrossings()
{
var inbound = ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiInbound, AuditKind.InboundRequest, AuditStatus.Delivered,
occurredAtUtc: T0, target: "M", sourceNode: "central-a", executionId: Root, durationMs: 50, responseSummary: "{}");
var centralDb = ScadaBridgeAuditEventFactory.Create(AuditChannel.DbOutbound, AuditKind.DbWrite, AuditStatus.Delivered,
occurredAtUtc: T0.AddMilliseconds(5), actor: "api:M", target: "CfgDb", sourceNode: "central-a", executionId: Root,
extra: "{\"op\":\"read\",\"rowsReturned\":2}");
var graph = FlowGraphBuilder.Build([inbound, centralDb]);
Assert.DoesNotContain(graph.Edges, e => e.CrossesBoundary);
Assert.Equal("read · 2 rows", graph.Steps[1].Subtitle);
// Response links from the DB call — the inbound execution's last sync step.
var into = Assert.Single(graph.Edges, e => e.ToEventId == graph.Steps[2].EventId);
Assert.Equal(graph.Steps[1].EventId, into.FromEventId);
}
}
tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Debugging/FlowStepProjectorTests.cs
+13
View File
@@ -148,4 +148,17 @@ public class FlowStepProjectorTests
Assert.Equal(FlowStepOutcome.Ok, FlowStepProjector.Project(ok).Outcome);
Assert.Equal(FlowStepOutcome.Failed, FlowStepProjector.Project(failed).Outcome);
}
[Theory]
[InlineData("{\"op\":\"write\",\"rowsAffected\":2147483648}", "write")]
[InlineData("{\"op\":\"write\",\"rowsAffected\":1.5}", "write")]
[InlineData("{\"op\":\"write\",\"rowsAffected\":-1}", "write")]
[InlineData("{\"op\":7}", "Delivered")]
[InlineData("not json", "Delivered")]
public void DbSubtitle_NeverThrows_OnOddExtra(string extra, string expected)
{
var evt = ScadaBridgeAuditEventFactory.Create(AuditChannel.DbOutbound, AuditKind.DbWrite, AuditStatus.Delivered,
occurredAtUtc: T0.UtcDateTime, target: "MesDb", extra: extra);
Assert.Equal(expected, FlowStepProjector.Project(evt).Subtitle);
}
}