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

Merge branch 'worktree-agent-a0f225cd55f5aa409' into fix/flow-debugger-arch-review

Browse Source
This commit is contained in:
Joseph Doherty
2026-09-06 14:13:08 -04:00
parent 17aea09fca 297ae56b85
commit 38509441bb
1 changed files with 166 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs
+166 -173
View File
@@ -1,4 +1,5 @@
using System.Data; using System.Data;
using System.Data.Common;
using System.Text; using System.Text;
using Microsoft.Data.SqlClient; using Microsoft.Data.SqlClient;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
@@ -918,105 +919,114 @@ VALUES
private static readonly TimeSpan ExecutionTreeWindowSlack = TimeSpan.FromHours(1); private static readonly TimeSpan ExecutionTreeWindowSlack = TimeSpan.FromHours(1);
private static readonly TimeSpan ExecutionTreeMaxSpan = TimeSpan.FromDays(7); private static readonly TimeSpan ExecutionTreeMaxSpan = TimeSpan.FromDays(7);
/// <inheritdoc /> /// <summary>
public async Task<IReadOnlyList<ExecutionTreeNode>> GetExecutionTreeAsync( /// The anchor of one execution chain: the root execution id reached by the upward
Guid executionId, /// <c>ParentExecutionId</c> walk, plus that root's first event timestamp (null when the
CancellationToken ct = default) /// root is a row-less stub). Produced by <see cref="ResolveChainAnchorAsync"/> and
/// consumed by <see cref="ChainCtePrefix"/> / <see cref="AddChainParameters"/>.
/// </summary>
private sealed record ChainAnchor(Guid RootExecutionId, DateTime? RootFirstOccurred)
{ {
var conn = _context.Database.GetDbConnection(); /// <summary>
var openedHere = false; /// The optional window predicate for the Edges CTE. When the root has rows the scan is
if (conn.State != System.Data.ConnectionState.Open) /// confined to the window (partition elimination); otherwise it stays unbounded.
/// </summary>
public string EdgesWindowPredicate => RootFirstOccurred is null
? string.Empty
: "AND OccurredAtUtc >= @winStart AND OccurredAtUtc < @winEnd";
}
/// <summary>
/// Phase 1 of the execution-chain walk, shared by <see cref="GetExecutionTreeAsync"/> and
/// <see cref="GetTraceEventsAsync"/> (arch-review F9): climb to the chain root, then read the
/// root's first event timestamp so the down-walk can be bounded to a root-anchored window.
/// The two walkers must stay in step on window slack, depth cap and the DISTINCT-edges trick,
/// so there is exactly one copy of each.
/// </summary>
private static async Task<ChainAnchor> ResolveChainAnchorAsync(
DbConnection conn,
Guid executionId,
CancellationToken ct)
{
// --- Phase 1: walk up to the root ---------------------------------
// Climb ParentExecutionId until a node has no parent (root) or the
// parent execution has no rows of its own (purged/stub — the climb
// cannot continue past a row-less node). The depth cap guards
// against a cycle in corrupt data; a tree never reaches it.
var rootExecutionId = executionId;
for (var depth = 0; depth < ExecutionChainMaxDepth; depth++)
{ {
await conn.OpenAsync(ct).ConfigureAwait(false); Guid? parent;
openedHere = true; await using (var upCmd = conn.CreateCommand())
{
upCmd.CommandText =
"SELECT TOP 1 ParentExecutionId FROM dbo.AuditLog " +
"WHERE ExecutionId = @cur AND ParentExecutionId IS NOT NULL;";
var pCur = upCmd.CreateParameter();
pCur.ParameterName = "@cur";
pCur.Value = rootExecutionId;
upCmd.Parameters.Add(pCur);
var result = await upCmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
parent = result is null or DBNull ? null : (Guid)result;
}
if (parent is null)
{
// No parent row for the current node — it is the root (or a
// row-less stub at the top of what survives). Stop climbing.
break;
}
rootExecutionId = parent.Value;
} }
try // --- Between phases: bound the down-walk to a root-anchored window ---
// The root's first event timestamp anchors a [rootFirst - slack, rootFirst + maxSpan)
// window used to prune the Edges CTE scan to a handful of partitions. A row-less stub
// root (purged/no-action parent) has no MIN — leave the scan unbounded in that
// degenerate case (correctness over speed).
DateTime? rootFirstOccurred;
await using (var winCmd = conn.CreateCommand())
{ {
// --- Phase 1: walk up to the root --------------------------------- winCmd.CommandText =
// Climb ParentExecutionId until a node has no parent (root) or the "SELECT MIN(OccurredAtUtc) FROM dbo.AuditLog WHERE ExecutionId = @root;";
// parent execution has no rows of its own (purged/stub — the climb var pWinRoot = winCmd.CreateParameter();
// cannot continue past a row-less node). The depth cap guards pWinRoot.ParameterName = "@root";
// against a cycle in corrupt data; a tree never reaches it. pWinRoot.Value = rootExecutionId;
var rootExecutionId = executionId; winCmd.Parameters.Add(pWinRoot);
for (var depth = 0; depth < ExecutionChainMaxDepth; depth++)
{
Guid? parent;
await using (var upCmd = conn.CreateCommand())
{
upCmd.CommandText =
"SELECT TOP 1 ParentExecutionId FROM dbo.AuditLog " +
"WHERE ExecutionId = @cur AND ParentExecutionId IS NOT NULL;";
var pCur = upCmd.CreateParameter();
pCur.ParameterName = "@cur";
pCur.Value = rootExecutionId;
upCmd.Parameters.Add(pCur);
var result = await upCmd.ExecuteScalarAsync(ct).ConfigureAwait(false); var winResult = await winCmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
parent = result is null or DBNull ? null : (Guid)result; rootFirstOccurred = winResult is null or DBNull ? null : (DateTime)winResult;
} }
if (parent is null) return new ChainAnchor(rootExecutionId, rootFirstOccurred);
{ }
// No parent row for the current node — it is the root (or a
// row-less stub at the top of what survives). Stop climbing.
break;
}
rootExecutionId = parent.Value; /// <summary>
} /// Phase 2's Edges/Chain CTE prefix — identical in both walkers (arch-review F9). Callers
/// append their own projection: <see cref="GetExecutionTreeAsync"/> the <c>ChainIds</c> CTE
// --- Between phases: bound the down-walk to a root-anchored window --- /// plus the per-execution aggregate, <see cref="GetTraceEventsAsync"/> the bare id union.
// The root's first event timestamp anchors a [rootFirst - slack, rootFirst + maxSpan) /// Both must terminate with <c>OPTION (MAXRECURSION …)</c>.
// window used to prune the Edges CTE scan to a handful of partitions. A row-less stub /// </summary>
// root (purged/no-action parent) has no MIN — leave the scan unbounded in that /// <remarks>
// degenerate case (correctness over speed). /// Edges : a non-recursive, DISTINCT (ExecutionId, ParentExecutionId)
DateTime? rootFirstOccurred; /// edge set distilled from AuditLog. Recursing over edges
await using (var winCmd = conn.CreateCommand()) /// instead of raw rows means an execution with N audit rows
{ /// contributes ONE recursion path, not N — MAXRECURSION
winCmd.CommandText = /// bounds depth, not per-level width, so the raw-row form
"SELECT MIN(OccurredAtUtc) FROM dbo.AuditLog WHERE ExecutionId = @root;"; /// could fan out badly. One edge per execution because all
var pWinRoot = winCmd.CreateParameter(); /// rows of an execution share a single ParentExecutionId
pWinRoot.ParameterName = "@root"; /// (see the MIN(...) note on the tree walker's projection).
pWinRoot.Value = rootExecutionId; /// Chain : seeded at the root edge, recursively joins each edge whose
winCmd.Parameters.Add(pWinRoot); /// ParentExecutionId is an ExecutionId already in the chain.
/// Each edge carries its own ParentExecutionId, so the chain
var winResult = await winCmd.ExecuteScalarAsync(ct).ConfigureAwait(false); /// of edges already surfaces every execution id in the tree
rootFirstOccurred = winResult is null or DBNull ? null : (DateTime)winResult; /// — including a row-less stub parent, which appears as the
} /// ParentExecutionId of its child's edge. No separate
/// union-back CTE is needed.
// Build the optional window predicate for the Edges CTE. When the root has rows the /// </remarks>
// scan is confined to the window (partition elimination); otherwise it stays unbounded. private static string ChainCtePrefix(string edgesWindowPredicate) => $@"
var edgesWindowPredicate = rootFirstOccurred is null
? string.Empty
: "AND OccurredAtUtc >= @winStart AND OccurredAtUtc < @winEnd";
// --- Phase 2: walk down from the root via a recursive CTE ---------
// Edges : a non-recursive, DISTINCT (ExecutionId, ParentExecutionId)
// edge set distilled from AuditLog. Recursing over edges
// instead of raw rows means an execution with N audit rows
// contributes ONE recursion path, not N — MAXRECURSION
// bounds depth, not per-level width, so the raw-row form
// could fan out badly. One edge per execution because all
// rows of an execution share a single ParentExecutionId
// (see the MIN(...) note on the final projection).
// Chain : seeded at the root edge, recursively joins each edge whose
// ParentExecutionId is an ExecutionId already in the chain.
// Each edge carries its own ParentExecutionId, so the chain
// of edges already surfaces every execution id in the tree
// — including a row-less stub parent, which appears as the
// ParentExecutionId of its child's edge. No separate
// union-back CTE is needed.
// Final : collect every distinct execution id reachable from the
// chain (each edge's ExecutionId plus its non-null
// ParentExecutionId), LEFT JOIN back to AuditLog and
// GROUP BY so a stub parent — which owns no edge of its own
// because it emitted no rows — still surfaces as a node with
// RowCount 0 and NULL aggregates.
var nodes = new List<ExecutionTreeNode>();
await using (var downCmd = conn.CreateCommand())
{
downCmd.CommandText = $@"
WITH Edges AS ( WITH Edges AS (
SELECT DISTINCT ExecutionId, ParentExecutionId SELECT DISTINCT ExecutionId, ParentExecutionId
FROM dbo.AuditLog FROM dbo.AuditLog
@@ -1035,7 +1045,65 @@ VALUES
SELECT e.ExecutionId, e.ParentExecutionId SELECT e.ExecutionId, e.ParentExecutionId
FROM Edges e FROM Edges e
INNER JOIN Chain c ON e.ParentExecutionId = c.ExecutionId INNER JOIN Chain c ON e.ParentExecutionId = c.ExecutionId
), )";
/// <summary>
/// Binds the parameters the <see cref="ChainCtePrefix"/> statement needs: <c>@root</c> always,
/// and the window bounds only when the anchor carries a root timestamp (matching
/// <see cref="ChainAnchor.EdgesWindowPredicate"/>).
/// </summary>
private static void AddChainParameters(DbCommand cmd, ChainAnchor anchor)
{
var pRoot = cmd.CreateParameter();
pRoot.ParameterName = "@root";
pRoot.Value = anchor.RootExecutionId;
cmd.Parameters.Add(pRoot);
if (anchor.RootFirstOccurred is { } rootFirst)
{
var pWinStart = cmd.CreateParameter();
pWinStart.ParameterName = "@winStart";
pWinStart.Value = rootFirst - ExecutionTreeWindowSlack;
cmd.Parameters.Add(pWinStart);
var pWinEnd = cmd.CreateParameter();
pWinEnd.ParameterName = "@winEnd";
pWinEnd.Value = rootFirst + ExecutionTreeMaxSpan;
cmd.Parameters.Add(pWinEnd);
}
}
/// <inheritdoc />
public async Task<IReadOnlyList<ExecutionTreeNode>> GetExecutionTreeAsync(
Guid executionId,
CancellationToken ct = default)
{
var conn = _context.Database.GetDbConnection();
var openedHere = false;
if (conn.State != System.Data.ConnectionState.Open)
{
await conn.OpenAsync(ct).ConfigureAwait(false);
openedHere = true;
}
try
{
// --- Phase 1 + window: shared with GetTraceEventsAsync (arch-review F9) ---
var anchor = await ResolveChainAnchorAsync(conn, executionId, ct).ConfigureAwait(false);
// --- Phase 2: walk down from the root via a recursive CTE ---------
// The Edges/Chain prefix is ChainCtePrefix (see its remarks for the
// DISTINCT-edges trick); this walker appends:
// ChainIds: collect every distinct execution id reachable from the
// chain (each edge's ExecutionId plus its non-null
// ParentExecutionId), LEFT JOIN back to AuditLog and
// GROUP BY so a stub parent — which owns no edge of its own
// because it emitted no rows — still surfaces as a node with
// RowCount 0 and NULL aggregates.
var nodes = new List<ExecutionTreeNode>();
await using (var downCmd = conn.CreateCommand())
{
downCmd.CommandText = ChainCtePrefix(anchor.EdgesWindowPredicate) + $@",
ChainIds AS ( ChainIds AS (
SELECT ExecutionId FROM Chain SELECT ExecutionId FROM Chain
UNION UNION
@@ -1071,23 +1139,7 @@ VALUES
GROUP BY ids.ExecutionId GROUP BY ids.ExecutionId
OPTION (MAXRECURSION {ExecutionChainMaxDepth});"; OPTION (MAXRECURSION {ExecutionChainMaxDepth});";
var pRoot = downCmd.CreateParameter(); AddChainParameters(downCmd, anchor);
pRoot.ParameterName = "@root";
pRoot.Value = rootExecutionId;
downCmd.Parameters.Add(pRoot);
if (rootFirstOccurred is { } rootFirst)
{
var pWinStart = downCmd.CreateParameter();
pWinStart.ParameterName = "@winStart";
pWinStart.Value = rootFirst - ExecutionTreeWindowSlack;
downCmd.Parameters.Add(pWinStart);
var pWinEnd = downCmd.CreateParameter();
pWinEnd.ParameterName = "@winEnd";
pWinEnd.Value = rootFirst + ExecutionTreeMaxSpan;
downCmd.Parameters.Add(pWinEnd);
}
await using var reader = await downCmd.ExecuteReaderAsync(ct).ConfigureAwait(false); await using var reader = await downCmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await reader.ReadAsync(ct).ConfigureAwait(false)) while (await reader.ReadAsync(ct).ConfigureAwait(false))
@@ -1147,77 +1199,18 @@ VALUES
var chainIds = new List<Guid>(); var chainIds = new List<Guid>();
try try
{ {
var rootExecutionId = executionId; var anchor = await ResolveChainAnchorAsync(conn, executionId, ct).ConfigureAwait(false);
for (var depth = 0; depth < ExecutionChainMaxDepth; depth++)
{
Guid? parent;
await using (var upCmd = conn.CreateCommand())
{
upCmd.CommandText =
"SELECT TOP 1 ParentExecutionId FROM dbo.AuditLog " +
"WHERE ExecutionId = @cur AND ParentExecutionId IS NOT NULL;";
var pCur = upCmd.CreateParameter();
pCur.ParameterName = "@cur";
pCur.Value = rootExecutionId;
upCmd.Parameters.Add(pCur);
var result = await upCmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
parent = result is null or DBNull ? null : (Guid)result;
}
if (parent is null) break;
rootExecutionId = parent.Value;
}
DateTime? rootFirstOccurred;
await using (var winCmd = conn.CreateCommand())
{
winCmd.CommandText = "SELECT MIN(OccurredAtUtc) FROM dbo.AuditLog WHERE ExecutionId = @root;";
var pWinRoot = winCmd.CreateParameter();
pWinRoot.ParameterName = "@root";
pWinRoot.Value = rootExecutionId;
winCmd.Parameters.Add(pWinRoot);
var winResult = await winCmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
rootFirstOccurred = winResult is null or DBNull ? null : (DateTime)winResult;
}
var edgesWindowPredicate = rootFirstOccurred is null
? string.Empty
: "AND OccurredAtUtc >= @winStart AND OccurredAtUtc < @winEnd";
await using (var idsCmd = conn.CreateCommand()) await using (var idsCmd = conn.CreateCommand())
{ {
idsCmd.CommandText = $@" idsCmd.CommandText = ChainCtePrefix(anchor.EdgesWindowPredicate) + $@"
WITH Edges AS (
SELECT DISTINCT ExecutionId, ParentExecutionId
FROM dbo.AuditLog
WHERE ExecutionId IS NOT NULL
{edgesWindowPredicate}
),
Chain AS (
SELECT CAST(@root AS uniqueidentifier) AS ExecutionId,
CAST(NULL AS uniqueidentifier) AS ParentExecutionId
UNION ALL
SELECT e.ExecutionId, e.ParentExecutionId
FROM Edges e
INNER JOIN Chain c ON e.ParentExecutionId = c.ExecutionId
)
SELECT ExecutionId FROM Chain SELECT ExecutionId FROM Chain
UNION UNION
SELECT ParentExecutionId FROM Chain WHERE ParentExecutionId IS NOT NULL SELECT ParentExecutionId FROM Chain WHERE ParentExecutionId IS NOT NULL
OPTION (MAXRECURSION {ExecutionChainMaxDepth});"; OPTION (MAXRECURSION {ExecutionChainMaxDepth});";
var pRoot = idsCmd.CreateParameter();
pRoot.ParameterName = "@root"; AddChainParameters(idsCmd, anchor);
pRoot.Value = rootExecutionId;
idsCmd.Parameters.Add(pRoot);
if (rootFirstOccurred is { } rootFirst)
{
var pWinStart = idsCmd.CreateParameter();
pWinStart.ParameterName = "@winStart";
pWinStart.Value = rootFirst - ExecutionTreeWindowSlack;
idsCmd.Parameters.Add(pWinStart);
var pWinEnd = idsCmd.CreateParameter();
pWinEnd.ParameterName = "@winEnd";
pWinEnd.Value = rootFirst + ExecutionTreeMaxSpan;
idsCmd.Parameters.Add(pWinEnd);
}
await using var reader = await idsCmd.ExecuteReaderAsync(ct).ConfigureAwait(false); await using var reader = await idsCmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await reader.ReadAsync(ct).ConfigureAwait(false)) while (await reader.ReadAsync(ct).ConfigureAwait(false))
{ {