diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Debugging/FlowGraphBuilder.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Debugging/FlowGraphBuilder.cs index 143064c0..96138463 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Debugging/FlowGraphBuilder.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Debugging/FlowGraphBuilder.cs @@ -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; /// DurationMs is the sync row's own, or submit→terminal wall time for a cached operation (null while pending); /// Lifecycle lists every row in time order. -/// Steps are ordered by OccurredAtUtc, then a kind rank (inbound before route call -/// before script run before writes/waits before outbound), then EventId. +/// Order = 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 OccurredAtUtc, 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. /// Within one ExecutionId consecutive steps are chained in that order, skipping async steps — a /// Notify, or a cached operation whose first row is CachedSubmit. Each skipped step instead gets an /// Async (dashed) edge from the chain step immediately before it in time. /// The first step of an execution is linked from its spawner: a RouteCall whose -/// childExecutionId matches, else the latest step of the parent execution that occurred -/// no later than it — that second case is a Spawn edge (trigger cascade). The edge +/// childExecutionId matches, else the last step of the parent execution that precedes it in +/// the order above — that second case is a Spawn edge (trigger cascade). The edge /// CrossesBoundary when exactly one end is in the Central lane. /// An AttributeWait that was satisfied gets an External (dashed) edge from the /// preceding AttributeWrite in the same execution — the device echo the system inferred. @@ -41,12 +44,47 @@ public static class FlowGraphBuilder if (events.Count == 0) return new FlowGraph(Array.Empty(), Array.Empty(), 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(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(steps.Count); + for (var i = 0; i < steps.Count; i++) + position[steps[i].EventId] = i; + var edges = new List(); 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; diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs index 449f246b..00470a04 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs @@ -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(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, diff --git a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Debugging/FlowGraphBuilderTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Debugging/FlowGraphBuilderTests.cs index 1bee9430..d2b4e15c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Debugging/FlowGraphBuilderTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Debugging/FlowGraphBuilderTests.cs @@ -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 ------------ + + /// + /// 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. + /// + private static List 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] diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs index b8f700c2..4595a45e 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs @@ -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()).Returns(method); + _locator.GetSiteIdForInstanceAsync("inst", Arg.Any()).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(), Arg.Any()) + .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() {