Resolve all 46 low-severity architecture review findings

This commit is contained in:
Joseph Doherty
2026-09-09 22:56:36 -04:00
parent fef3e7675f
commit 9ee402bbcd
139 changed files with 2621 additions and 1082 deletions
@@ -292,7 +292,8 @@ public class AuditLogPurgeActor : ReceiveActor
channel,
channelThreshold,
_purgeOptions.ChannelPurgeBatchSize,
_purgeOptions.ResolvedMaintenanceCommandTimeout)
_purgeOptions.ResolvedMaintenanceCommandTimeout,
maxRows: _purgeOptions.MaxPurgedRowsPerChannelPerTick)
.ConfigureAwait(false);
sw.Stop();
@@ -30,13 +30,16 @@ public sealed class AuditLogPurgeOptions
/// <summary>Period of the purge tick in hours (default 24).</summary>
public int IntervalHours { get; set; } = 24;
/// <summary>Maximum rows deleted for each channel in one purge tick (default 50,000); remaining backlog resumes on later ticks.</summary>
public int MaxPurgedRowsPerChannelPerTick { get; set; } = 50_000;
/// <summary>
/// Batch size for the per-channel retention-override row DELETE
/// (<see cref="ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories.IAuditLogRepository.PurgeChannelOlderThanAsync"/>).
/// Each <c>DELETE TOP (@batch)</c> caps the transaction-log and lock footprint
/// per statement; the repository loops batches until no rows remain. Default
/// 5000 keeps individual deletes short on a busy central DB while still draining
/// a large backlog within a tick. Clamped to a sane minimum in
/// per statement; the repository stops at <see cref="MaxPurgedRowsPerChannelPerTick"/>
/// so a backlogged channel yields to later channels. Default 5000 keeps individual
/// deletes short on a busy central DB. Clamped to a sane minimum in
/// <see cref="ChannelPurgeBatchSize"/>.
/// </summary>
/// <remarks>
@@ -19,6 +19,9 @@ public sealed class AuditLogPurgeOptionsValidator : OptionsValidatorBase<AuditLo
$"AuditLog:Purge:{nameof(AuditLogPurgeOptions.IntervalHours)} " +
$"({options.IntervalHours}) must be > 0.");
builder.RequireThat(options.MaxPurgedRowsPerChannelPerTick > 0,
"AuditLog:Purge:MaxPurgedRowsPerChannelPerTick must be > 0.");
// Backing field for the documented "ChannelPurgeBatchSize" config key.
builder.RequireThat(options.ChannelPurgeBatchSizeConfigured > 0,
$"AuditLog:Purge:{nameof(AuditLogPurgeOptions.ChannelPurgeBatchSizeConfigured)} " +
@@ -44,7 +44,7 @@ public sealed class GrpcPullTraceEventsClient : IPullTraceEventsClient
IReadOnlyCollection<Guid> executionIds,
DateTime sinceUtc,
int maxEvents,
CancellationToken ct)
CancellationToken ct, string? afterEventId = null)
{
ArgumentNullException.ThrowIfNull(executionIds);
if (executionIds.Count == 0 || maxEvents <= 0)
@@ -67,6 +67,7 @@ public sealed class GrpcPullTraceEventsClient : IPullTraceEventsClient
{
SinceUtc = Timestamp.FromDateTime(EnsureUtc(sinceUtc)),
MaxEvents = maxEvents,
AfterId = afterEventId ?? string.Empty,
};
foreach (var id in executionIds)
{
@@ -20,9 +20,9 @@ public sealed class SiteAuditReconciliationOptionsValidator
$"AuditLog:Reconciliation:{nameof(SiteAuditReconciliationOptions.ReconciliationIntervalSeconds)} " +
$"({options.ReconciliationIntervalSeconds}) must be > 0.");
builder.RequireThat(options.BatchSize > 0,
builder.RequireThat(options.BatchSize is > 0 and <= Communication.Grpc.SiteStreamGrpcServer.MaxReconciliationRowsPerPull,
$"AuditLog:Reconciliation:{nameof(SiteAuditReconciliationOptions.BatchSize)} " +
$"({options.BatchSize}) must be > 0.");
$"({options.BatchSize}) must be 1..2000.");
builder.RequireThat(options.StalledAfterNonDrainingCycles > 0,
$"AuditLog:Reconciliation:{nameof(SiteAuditReconciliationOptions.StalledAfterNonDrainingCycles)} " +
@@ -271,6 +271,15 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
-- to the window. Additive and idempotent.
CREATE INDEX IF NOT EXISTS IX_audit_event_occurred
ON audit_event (OccurredAtUtc);
-- Fresh execution IDs permit clock-independent trace reads. Partial indexes
-- exclude malformed legacy details before evaluating either JSON expression.
CREATE INDEX IF NOT EXISTS IX_audit_event_execution
ON audit_event (json_extract(DetailsJson, '$.executionId'), OccurredAtUtc, EventId)
WHERE json_valid(DetailsJson);
CREATE INDEX IF NOT EXISTS IX_audit_event_parent_execution
ON audit_event (json_extract(DetailsJson, '$.parentExecutionId'), OccurredAtUtc, EventId)
WHERE json_valid(DetailsJson);
""";
cmd.ExecuteNonQuery();
}
@@ -859,7 +868,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
/// <inheritdoc />
public Task<IReadOnlyList<AuditEvent>> ReadByExecutionIdsAsync(
IReadOnlyCollection<Guid> executionIds, DateTime sinceUtc, int limit, CancellationToken ct = default)
IReadOnlyCollection<Guid> executionIds, DateTime sinceUtc, int limit, CancellationToken ct = default, string? afterEventId = null)
{
ArgumentNullException.ThrowIfNull(executionIds);
if (limit <= 0)
@@ -895,18 +904,10 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
}
var inList = string.Join(", ", names);
cmd.CommandText = $"""
SELECT EventId, OccurredAtUtc, Actor, Action, Outcome,
Category, Target, SourceNode, CorrelationId, DetailsJson
FROM audit_event
WHERE OccurredAtUtc >= $since
AND (json_extract(DetailsJson, '$.executionId') IN ({inList})
OR json_extract(DetailsJson, '$.parentExecutionId') IN ({inList}))
ORDER BY OccurredAtUtc ASC, EventId ASC
LIMIT $limit;
""";
cmd.CommandText = BuildTraceReadSql(inList);
cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString(
"o", System.Globalization.CultureInfo.InvariantCulture));
cmd.Parameters.AddWithValue("$after", (object?)afterEventId ?? DBNull.Value);
cmd.Parameters.AddWithValue("$limit", limit);
var rows = new List<AuditEvent>(Math.Min(limit, 256));
@@ -919,6 +920,33 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
}
}
/// <summary>
/// Reads matching keys through both expression indexes before loading the bounded payload page.
/// UNION deduplicates rows matching both IDs. Keeping payloads outside the union bounds sorting
/// to IDs/timestamps and avoids the optimizer preferring a whole-store timestamp scan.
/// </summary>
internal static string BuildTraceReadSql(string inList) => $"""
SELECT e.EventId, e.OccurredAtUtc, e.Actor, e.Action, e.Outcome,
e.Category, e.Target, e.SourceNode, e.CorrelationId, e.DetailsJson
FROM audit_event AS e
JOIN (
SELECT EventId, OccurredAtUtc
FROM audit_event INDEXED BY IX_audit_event_execution
WHERE json_valid(DetailsJson)
AND json_extract(DetailsJson, '$.executionId') IN ({inList})
AND (OccurredAtUtc > $since OR (OccurredAtUtc = $since AND ($after IS NULL OR EventId > $after)))
UNION
SELECT EventId, OccurredAtUtc
FROM audit_event INDEXED BY IX_audit_event_parent_execution
WHERE json_valid(DetailsJson)
AND json_extract(DetailsJson, '$.parentExecutionId') IN ({inList})
AND (OccurredAtUtc > $since OR (OccurredAtUtc = $since AND ($after IS NULL OR EventId > $after)))
ORDER BY OccurredAtUtc, EventId
LIMIT $limit
) AS matched ON matched.EventId = e.EventId
ORDER BY matched.OccurredAtUtc, matched.EventId;
""";
/// <summary>
/// Highest <c>audit_forward_state.rowid</c> this instance has ever SERVED from
/// <see cref="ReadPendingSinceAsync"/> — i.e. the insertion-order high-water mark of
+10
View File
@@ -2270,6 +2270,16 @@ scadabridge --url <url> cached-call discard --site-id <string> --tracked-operati
---
### `secured-write` — Query two-person write requests
`secured-write list` reads the MxGateway secured-write queue and terminal history. It requires any of Operator, Verifier or Administrator and applies the caller's site scope. This CLI group exposes listing only; submitting, approving and rejecting are management/UI operations.
```sh
scadabridge secured-write list --status Pending --site site-a --skip 0 --take 50
```
`--status` and `--site` are optional filters. `--skip` defaults to 0; `--take` defaults to 200 and is clamped server-side to 1–500. Global connection, authentication and output-format options apply.
### `flow-debug` — Trace an inbound API, script, or value-change flow end to end
The CLI face of the **Flow Debugger** (design `docs/plans/2026-09-05-flow-debugger-design.md`,
@@ -25,6 +25,13 @@ public static class AuthEndpoints
{
endpoints.MapPost("/auth/login", async (HttpContext context) =>
{
var antiforgery = context.RequestServices.GetRequiredService<IAntiforgery>();
try { await antiforgery.ValidateRequestAsync(context); }
catch (AntiforgeryValidationException)
{
context.Response.StatusCode = StatusCodes.Status400BadRequest;
return;
}
var form = await context.Request.ReadFormAsync();
var username = form["username"].ToString();
var password = form["password"].ToString();
@@ -109,7 +116,7 @@ public static class AuthEndpoints
BuildSignInProperties());
context.Response.Redirect("/");
}).DisableAntiforgery();
});
endpoints.MapPost("/auth/token", async (HttpContext context) =>
{
@@ -1,10 +1,9 @@
@using ZB.MOM.WW.ScadaBridge.CentralUI.Auth
@inject Microsoft.AspNetCore.Components.Authorization.AuthenticationStateProvider MutationAuth
@using Microsoft.AspNetCore.Components.Authorization
@using ZB.MOM.WW.ScadaBridge.CentralUI.Components.Shared
@using ZB.MOM.WW.ScadaBridge.CentralUI.Services
@using ZB.MOM.WW.ScadaBridge.Security
@inject IManualFailoverService Failover
@inject AuthorizedManualFailoverService Failover
@inject IDialogService Dialog
@* Admin-only manual failover of the central pair (decision 2026-07-22). Rendered on the
@@ -94,11 +93,10 @@
try
{
var actor = await MutationAuth.RequireRoleAsync(ZB.MOM.WW.ScadaBridge.Security.Roles.Administrator);
if (IsSite)
{
var outcome = await Failover.FailOverSiteAsync(SiteId!, actor);
var outcome = await Failover.FailOverSiteAsync(SiteId!);
if (outcome.Accepted)
{
_message = $"Failover triggered — {outcome.TargetAddress} is leaving the site cluster.";
@@ -113,7 +111,7 @@
}
else
{
var target = await Failover.FailOverCentralAsync(actor);
var target = await Failover.FailOverCentralAsync();
if (target is null)
{
// The server-side peer guard refused. Never report a failover that did not happen.
@@ -47,33 +47,38 @@
<NavRailItem Href="/deployment/topology" Text="Topology" />
<NavRailItem Href="/deployment/deployments" Text="Deployments" />
<NavRailItem Href="/deployment/debug-view" Text="Debug View" />
<NavRailItem Href="/deployment/flow-debugger" Text="Flow Debugger" />
</NavRailSection>
</Authorized>
</AuthorizeView>
@* Notifications — mixed-role section; each item gated by its own policy.
The section is ungated: every authenticated user holds at least one of
Admin/Design/Deployment, so it always has a visible child. *@
<NavRailSection Title="Notifications" Key="notifications">
<AuthorizeView Policy="@AuthorizationPolicies.RequireAdmin">
<Authorized Context="notifAdminContext">
<NavRailItem Href="/notifications/smtp" Text="SMTP Configuration" />
<NavRailItem Href="/notifications/sms" Text="SMS Configuration" />
</Authorized>
</AuthorizeView>
<AuthorizeView Policy="@AuthorizationPolicies.RequireDesign">
<Authorized Context="notifDesignContext">
<NavRailItem Href="/notifications/lists" Text="Notification Lists" />
</Authorized>
</AuthorizeView>
<AuthorizeView Policy="@AuthorizationPolicies.RequireDeployment">
<Authorized Context="notifDeploymentContext">
<NavRailItem Href="/notifications/report" Text="Notification Report" />
<NavRailItem Href="/notifications/kpis" Text="Notification KPIs" />
</Authorized>
</AuthorizeView>
</NavRailSection>
<AuthorizeView Policy="@AuthorizationPolicies.RequireFlowDebugger">
<Authorized Context="flowContext">
<NavRailItem Href="/deployment/flow-debugger" Text="Flow Debugger" />
</Authorized>
</AuthorizeView>
@if (context.User.Claims.Any(c => c.Type == JwtTokenService.RoleClaimType && (c.Value == Roles.Administrator || c.Value == Roles.Designer || c.Value == Roles.Deployer)))
{
<NavRailSection Title="Notifications" Key="notifications">
<AuthorizeView Policy="@AuthorizationPolicies.RequireAdmin">
<Authorized Context="notifAdminContext">
<NavRailItem Href="/notifications/smtp" Text="SMTP Configuration" />
<NavRailItem Href="/notifications/sms" Text="SMS Configuration" />
</Authorized>
</AuthorizeView>
<AuthorizeView Policy="@AuthorizationPolicies.RequireDesign">
<Authorized Context="notifDesignContext">
<NavRailItem Href="/notifications/lists" Text="Notification Lists" />
</Authorized>
</AuthorizeView>
<AuthorizeView Policy="@AuthorizationPolicies.RequireDeployment">
<Authorized Context="notifDeploymentContext">
<NavRailItem Href="/notifications/report" Text="Notification Report" />
<NavRailItem Href="/notifications/kpis" Text="Notification KPIs" />
</Authorized>
</AuthorizeView>
</NavRailSection>
}
@* Site Calls — Site Call Audit (#22). Deployer-role only,
matching the Notification Report page's gate; the whole
@@ -704,12 +704,12 @@ public partial class FlowDebugger : ComponentBase, IDisposable
private static int CountLogRows(IReadOnlyList<AuditEvent>? events)
=> events?.Count(FlowStepProjector.IsLogLine) ?? 0;
/// <summary>Union of two audit-row lists by <see cref="AuditEvent.EventId"/>; a row present in both takes the fresh copy.</summary>
/// <summary>Preserves fresh Replay order, then appends live-only rows; duplicate event IDs use the fresh copy.</summary>
internal static IReadOnlyList<AuditEvent> MergeEvents(IReadOnlyList<AuditEvent>? known, IReadOnlyList<AuditEvent>? fresh)
{
var byId = new Dictionary<Guid, AuditEvent>();
foreach (var e in known ?? []) byId[e.EventId] = e;
foreach (var e in fresh ?? []) byId[e.EventId] = e;
foreach (var e in known ?? []) byId.TryAdd(e.EventId, e);
return byId.Values.ToList();
}
@@ -811,11 +811,11 @@ public partial class FlowDebugger : ComponentBase, IDisposable
/// <summary>
/// Server-side re-check before relaying — the action must not fire for a site outside the caller's
/// permitted set even though the node is on screen. Notifications are central-owned (no site).
/// A site the page cannot resolve is allowed through, as on the Site Calls page.
/// An unresolvable site is denied for a scoped caller.
/// </summary>
private async Task<bool> IsStepSiteAllowedAsync(FlowStep step)
{
if (step.Kind == FlowStepKind.Notify || string.IsNullOrEmpty(step.SiteId))
if (step.Kind == FlowStepKind.Notify)
return true;
if (await SiteScope.IsSystemWideAsync())
return true;
@@ -832,7 +832,7 @@ public partial class FlowDebugger : ComponentBase, IDisposable
}
}
if (site is null)
return true;
return false;
return (await SiteScope.PermittedSiteIdsAsync()).Contains(site.Id);
}
@@ -73,7 +73,7 @@
</div>
<div class="d-flex align-items-center mb-2 gap-2 flex-wrap">
<input type="text" class="form-control form-control-sm" style="max-width: 320px;"
placeholder="Search sites, areas, instances..."
placeholder="Search loaded sites, areas, instances..."
@bind="_searchText" @bind:event="oninput" @bind:after="OnSearchChanged" />
<div class="btn-group btn-group-sm" role="group" aria-label="Topology actions">
<button class="btn btn-outline-secondary" @onclick="OpenCreateAreaDialogRoot">+ Area</button>
@@ -84,7 +84,7 @@
<button class="btn btn-outline-secondary" aria-label="Collapse all areas" @onclick="() => _tree?.CollapseAll()">Collapse</button>
</div>
<button class="btn btn-outline-secondary btn-sm" aria-label="Re-check staleness"
title="Recompute the Stale/Current badge for every deployed instance"
title="Recompute the Stale/Current badge for loaded deployed instances"
@onclick="RecheckStalenessAsync" disabled="@_recheckingStaleness">
@if (_recheckingStaleness)
{
@@ -104,7 +104,7 @@
</div>
<div class="alert alert-light py-2 mb-3 small">
@_allAreas.Count area(s) · @_allInstances.Count instance(s) across @_sites.Count site(s).
@_allAreas.Count loaded area(s) · @_allInstances.Count loaded instance(s) across @_sites.Count site(s). Expand a branch and load its instances (50 per page).
</div>
<div style="max-height: calc(100vh - 240px); overflow-y: auto; padding: 4px;">
@@ -136,7 +136,11 @@
// ---- Data ----
private List<Instance> _allInstances = new();
private List<Site> _sites = new();
private List<Template> _templates = new();
private const int InstancePageSize = 50;
private readonly Dictionary<(int Site, int? Area), int> _loadedPages = new();
private readonly HashSet<int> _loadedSites = new();
private bool _branchLoading;
private readonly SemaphoreSlim _loadGate = new(1, 1);
private List<Area> _allAreas = new();
private Dictionary<int, bool> _stalenessMap = new();
@@ -183,7 +187,7 @@
private const string SelectedKeyStorage = "topology-tree-selected";
// ---- Tree model ----
private enum TopoNodeKind { Site, Area, Instance }
private enum TopoNodeKind { Site, Area, Instance, Loader }
private record TopoNode(
string Key,
@@ -330,45 +334,81 @@
}
/// <summary>
/// Fast path — loads the cheap deployed state (sites, instances, areas →
/// hierarchy + State badge) WITHOUT recomputing per-instance staleness. These
/// are all single, indexed central-DB reads. This is exactly what the
/// live-update timer polls every 15s. Internal so the live-update regression
/// Refreshes permitted sites and previously loaded branch pages without expanding
/// new sites or loading every instance. Each instance query is capped at 50 rows.
/// This is the live-update timer path; it does not recompute staleness. Internal so the live-update regression
/// test can drive the timer's code path directly.
/// </summary>
internal async Task LoadDeployedStateAsync()
{
_loading = true;
_errorMessage = null;
await _loadGate.WaitAsync();
try
{
// Site scoping (CentralUI-002): a scoped Deployment user only sees the
// sites — and therefore the areas/instances — they are permitted on.
_sites = await SiteScope.FilterSitesAsync(await SiteRepository.GetAllSitesAsync());
var permittedSiteIds = _sites.Select(s => s.Id).ToHashSet();
_allInstances = (await TemplateEngineRepository.GetAllInstancesAsync())
.Where(i => permittedSiteIds.Contains(i.SiteId))
.ToList();
_templates = (await TemplateEngineRepository.GetAllTemplatesAsync()).ToList();
_allAreas.Clear();
foreach (var site in _sites)
_loading = true;
_errorMessage = null;
try
{
var areas = await TemplateEngineRepository.GetAreasBySiteIdAsync(site.Id);
_allAreas.AddRange(areas);
}
// Site scoping (CentralUI-002): a scoped Deployment user only sees the
// sites — and therefore the areas/instances — they are permitted on.
_sites = await SiteScope.FilterSitesAsync(await SiteRepository.GetAllSitesAsync());
var permittedSiteIds = _sites.Select(s => s.Id).ToHashSet();
_loadedSites.IntersectWith(permittedSiteIds);
foreach (var key in _loadedPages.Keys.Where(k => !permittedSiteIds.Contains(k.Site)).ToArray())
_loadedPages.Remove(key);
_allInstances.Clear();
_allAreas.Clear();
foreach (var siteId in _loadedSites)
_allAreas.AddRange(await TemplateEngineRepository.GetAreasBySiteIdAsync(siteId));
foreach (var (key, offset) in _loadedPages)
_allInstances.AddRange(await TemplateEngineRepository.GetTopologyInstancesAsync(key.Site, key.Area, offset, InstancePageSize));
BuildTree();
}
catch (Exception ex)
{
_errorMessage = $"Failed to load topology: {ex.Message}";
}
_loading = false;
BuildTree();
}
catch (Exception ex)
{
_errorMessage = $"Failed to load topology: {ex.Message}";
}
_loading = false;
finally { _loadGate.Release(); }
}
// Explicit branch loading keeps collapsed sites/areas out of circuit memory.
internal async Task LoadBranchAsync(int siteId, int? areaId, int offset = 0)
{
await _loadGate.WaitAsync();
try
{
if (!await SiteScope.IsSiteAllowedAsync(siteId)) return;
_branchLoading = true;
try
{
if (!_loadedSites.Contains(siteId))
{
_allAreas.AddRange(await TemplateEngineRepository.GetAreasBySiteIdAsync(siteId));
_loadedSites.Add(siteId);
}
var rows = await TemplateEngineRepository.GetTopologyInstancesAsync(siteId, areaId, offset, InstancePageSize);
_allInstances.RemoveAll(i => i.SiteId == siteId && i.AreaId == areaId);
_allInstances.AddRange(rows);
_loadedPages[(siteId, areaId)] = offset;
await ComputeStalenessAsync(loadGateHeld: true);
BuildTree();
}
catch (Exception ex) { _errorMessage = $"Failed to load branch: {ex.Message}"; }
finally { _branchLoading = false; }
}
finally { _loadGate.Release(); }
}
private TopoNode MakeLoader(int siteId, int? areaId) => new(
$"load:{siteId}:{areaId}", TopoNodeKind.Loader, areaId ?? 0, siteId,
"Load instances", null, null, null, false, false, new());
/// <summary>
/// Expensive path — recomputes the Stale/Current badge for every deployed
/// Expensive path — recomputes the Stale/Current badge for each loaded deployed
/// instance via <c>DeploymentService.GetDeploymentComparisonAsync</c>, which
/// re-flattens each instance's current config and compares it to the deployed
/// snapshot. This is the page's dominant cost, so it runs only on initial
@@ -378,27 +418,33 @@
/// DI scope so the parallel flattens never share the circuit-scoped DbContext
/// (mirrors KpiHistoryQueryService's per-query scope).
/// </summary>
private async Task ComputeStalenessAsync()
private async Task ComputeStalenessAsync(bool loadGateHeld = false)
{
var deployedIds = _allInstances
.Where(i => i.State != InstanceState.NotDeployed)
.Select(i => i.Id)
.ToList();
if (deployedIds.Count == 0)
if (!loadGateHeld) await _loadGate.WaitAsync();
try
{
_stalenessMap = new Dictionary<int, bool>();
var deployedIds = _allInstances
.Where(i => i.State != InstanceState.NotDeployed)
.Select(i => i.Id)
.ToList();
if (deployedIds.Count == 0)
{
_stalenessMap = new Dictionary<int, bool>();
BuildTree();
return;
}
var results = new ConcurrentDictionary<int, bool>();
using var gate = new SemaphoreSlim(MaxConcurrentStalenessChecks, MaxConcurrentStalenessChecks);
await Task.WhenAll(deployedIds.Select(id => ComputeOneStalenessAsync(id, gate, results)));
_stalenessMap = new Dictionary<int, bool>(results);
BuildTree();
return;
}
var results = new ConcurrentDictionary<int, bool>();
using var gate = new SemaphoreSlim(MaxConcurrentStalenessChecks, MaxConcurrentStalenessChecks);
await Task.WhenAll(deployedIds.Select(id => ComputeOneStalenessAsync(id, gate, results)));
_stalenessMap = new Dictionary<int, bool>(results);
BuildTree();
finally { if (!loadGateHeld) _loadGate.Release(); }
}
private async Task ComputeOneStalenessAsync(
@@ -469,7 +515,7 @@
.Select(MakeInstanceNode)
.ToList();
var children = areaChildren.Concat(rootInstances).ToList();
var children = areaChildren.Concat(rootInstances).Append(MakeLoader(site.Id, null)).ToList();
return new TopoNode(
Key: $"s:{site.Id}",
@@ -500,7 +546,7 @@
.OrderBy(i => i.UniqueName, StringComparer.OrdinalIgnoreCase)
.Select(MakeInstanceNode)
.ToList();
var children = childAreas.Concat(areaInstances).ToList();
var children = childAreas.Concat(areaInstances).Append(MakeLoader(area.SiteId, area.Id)).ToList();
return new TopoNode(
Key: $"a:{area.Id}",
Kind: TopoNodeKind.Area,
@@ -538,6 +584,21 @@
switch (node.Kind)
{
case TopoNodeKind.Loader:
var areaId = node.EntityId == 0 ? (int?)null : node.EntityId;
var loaded = _loadedPages.TryGetValue((node.SiteId, areaId), out var offset);
@if (!loaded)
{
<button class="btn btn-link btn-sm" disabled="@_branchLoading" @onclick="() => LoadBranchAsync(node.SiteId, areaId)">Load instances</button>
}
else
{
<span class="small text-muted">Page @(offset / InstancePageSize + 1)</span>
<button class="btn btn-link btn-sm" disabled="@(_branchLoading || offset == 0)" @onclick="() => LoadBranchAsync(node.SiteId, areaId, Math.Max(0, offset - InstancePageSize))">Previous</button>
<button class="btn btn-link btn-sm" disabled="@(_branchLoading || _allInstances.Count(i => i.SiteId == node.SiteId && i.AreaId == areaId) < InstancePageSize)" @onclick="() => LoadBranchAsync(node.SiteId, areaId, offset + InstancePageSize)">Next</button>
}
break;
case TopoNodeKind.Site:
<span class="tv-glyph" style="@labelStyle"><i class="bi bi-building"></i></span>
<span class="tv-label fw-semibold" style="@labelStyle" title="@node.Label">@node.Label</span>
@@ -396,19 +396,12 @@
StopTimer();
_refreshTimer = new Timer(_ =>
{
if (!_pollGate.TryEnter()) return;
InvokeAsync(async () =>
_ = _pollGate.RunAsync(() => InvokeAsync(async () =>
{
try
{
await RefreshAsync();
StateHasChanged();
}
finally
{
_pollGate.Exit();
}
});
if (_disposed) return;
await RefreshAsync();
StateHasChanged();
}), error => System.Diagnostics.Trace.TraceWarning("AlarmSummary refresh failed: {0}", error));
}, null, TimeSpan.FromSeconds(_autoRefreshSeconds), TimeSpan.FromSeconds(_autoRefreshSeconds));
}
@@ -718,19 +718,12 @@
_refreshTimer = new Timer(_ =>
{
if (!_pollGate.TryEnter()) return;
InvokeAsync(async () =>
_ = _pollGate.RunAsync(() => InvokeAsync(async () =>
{
try
{
await RefreshAllAsync(forceRefresh: false);
StateHasChanged();
}
finally
{
_pollGate.Exit();
}
});
if (_disposed) return;
await RefreshAllAsync(forceRefresh: false);
StateHasChanged();
}), error => System.Diagnostics.Trace.TraceWarning("Health refresh failed: {0}", error));
}, null, TimeSpan.FromSeconds(_autoRefreshSeconds), TimeSpan.FromSeconds(_autoRefreshSeconds));
}
@@ -996,8 +989,11 @@
_ => "bg-secondary"
};
private volatile bool _disposed;
public void Dispose()
{
_disposed = true;
_refreshTimer?.Dispose();
}
}
@@ -96,12 +96,47 @@
private HashSet<string>? _filterRevealed;
private Func<TemplateTreeNode, bool>? _initiallyExpanded;
// Compare immutable value snapshots, rather than entity/list reference identity:
// callers may rename, move, or edit composition children in place.
private sealed record TreeInput(string Kind, int Id, int? ParentId, string Name, bool Derived = false, bool Inherited = false, string? KeyOverride = null);
private List<TreeInput>? _treeInputs;
private Dictionary<int, IReadOnlyList<TemplateTreeNode>> _extraChildren = new();
protected override void OnParametersSet()
{
BuildTree();
var inputs = new List<TreeInput>();
foreach (var folder in Folders)
inputs.Add(new("folder", folder.Id, folder.ParentFolderId, folder.Name));
_extraChildren = new();
foreach (var template in Templates)
{
inputs.Add(new("template", template.Id, template.FolderId, template.Name, template.IsDerived));
if (!template.IsDerived && ExtraTemplateChildren is not null)
{
var children = ExtraTemplateChildren(template);
_extraChildren[template.Id] = children;
CaptureChildren(children, inputs);
}
}
if (_treeInputs is null || !inputs.SequenceEqual(_treeInputs))
{
_treeInputs = inputs;
BuildTree();
}
// Filter and RevealKeys are independent and may also be changed in place.
ApplyFilter();
}
private static void CaptureChildren(IReadOnlyList<TemplateTreeNode> children, List<TreeInput> inputs)
{
foreach (var child in children)
{
inputs.Add(new(child.Kind.ToString(), child.Id, null, child.Name, Inherited: child.IsInherited, KeyOverride: child.KeyOverride));
CaptureChildren(child.Children, inputs);
inputs.Add(new("end", 0, null, ""));
}
}
private void BuildTree()
{
var folderNodes = Folders.ToDictionary(
@@ -136,7 +171,7 @@
if (ExtraTemplateChildren != null)
{
foreach (var extra in ExtraTemplateChildren(t))
foreach (var extra in _extraChildren[t.Id])
{
tNode.Children.Add(extra);
}
@@ -231,6 +266,8 @@
Kind = node.Kind,
Id = node.Id,
Name = node.Name,
IsInherited = node.IsInherited,
KeyOverride = node.KeyOverride,
};
foreach (var k in keptChildren) clone.Children.Add(k);
if (keptChildren.Count > 0)
@@ -37,6 +37,7 @@ public static class ServiceCollectionExtensions
// Resolves the current user's permitted site set from their SiteId claims
// so Deployment/Monitoring pages can enforce site scoping.
services.AddScoped<SiteScopeService>();
services.AddScoped<AuthorizedManualFailoverService>();
services.AddScoped<ConfigurationCommandService>();
// Centralised dialog service: pages inject IDialogService and a single
@@ -0,0 +1,17 @@
using Microsoft.AspNetCore.Components.Authorization;
using ZB.MOM.WW.ScadaBridge.CentralUI.Auth;
using ZB.MOM.WW.ScadaBridge.Security;
namespace ZB.MOM.WW.ScadaBridge.CentralUI.Services;
/// <summary>Scoped application boundary that checks current circuit authority before host failover.</summary>
public sealed class AuthorizedManualFailoverService(AuthenticationStateProvider auth, IManualFailoverService backend)
{
/// <summary>Fails over central only for a currently authorized administrator.</summary>
public async Task<string?> FailOverCentralAsync()
=> await backend.FailOverCentralAsync(await auth.RequireRoleAsync(Roles.Administrator));
/// <summary>Fails over a site only for a currently authorized administrator.</summary>
public async Task<SiteFailoverOutcome> FailOverSiteAsync(string siteId)
=> await backend.FailOverSiteAsync(siteId, await auth.RequireRoleAsync(Roles.Administrator));
}
@@ -84,6 +84,7 @@ public sealed class FlowDebugService : IFlowDebugService
private readonly ManagementActorHolder _holder;
private readonly AuthenticationStateProvider _auth;
private readonly IInboundApiRepository _inboundApi;
private readonly ISiteRepository _sites;
private readonly ITemplateEngineRepository _templates;
private readonly IFlatteningPipeline _pipeline;
private readonly CommunicationService _comms;
@@ -116,8 +117,10 @@ public sealed class FlowDebugService : IFlowDebugService
ITemplateEngineRepository templates,
IFlatteningPipeline pipeline,
CommunicationService comms,
ILogger<FlowDebugService> logger)
ILogger<FlowDebugService> logger,
ISiteRepository sites)
{
_sites = sites ?? throw new ArgumentNullException(nameof(sites));
_holder = holder ?? throw new ArgumentNullException(nameof(holder));
_auth = auth ?? throw new ArgumentNullException(nameof(auth));
_inboundApi = inboundApi ?? throw new ArgumentNullException(nameof(inboundApi));
@@ -322,6 +325,13 @@ public sealed class FlowDebugService : IFlowDebugService
}
var site = step.SiteId!;
var scope = new SiteScopeService(_auth);
if (!await scope.IsSystemWideAsync())
{
var owner = (await _sites.GetAllSitesAsync(ct)).FirstOrDefault(s => s.SiteIdentifier == site);
if (owner is null || !await scope.IsSiteAllowedAsync(owner.Id))
return new(false, "You are not permitted to act on cached calls for this site.");
}
SiteCallRelayOutcome outcome;
bool siteReachable;
string? errorMessage;
@@ -12,6 +12,15 @@ public sealed class PollGate
/// may proceed; <see langword="false"/> if a previous refresh is still running.</returns>
public bool TryEnter() => Interlocked.CompareExchange(ref _inFlight, 1, 0) == 0;
/// <summary>Runs and observes a posted refresh, releasing the gate even if dispatch is rejected.</summary>
public async Task RunAsync(Func<Task> dispatch, Action<Exception> onError)
{
if (!TryEnter()) return;
try { await dispatch().ConfigureAwait(false); }
catch (Exception error) { onError(error); }
finally { Exit(); }
}
/// <summary>Releases the gate after a poll tick's refresh has completed.</summary>
public void Exit() => Volatile.Write(ref _inFlight, 0);
}
@@ -168,8 +168,8 @@ public interface IAuditLogRepository
/// <para>
/// <b>Bounded + idempotent.</b> Each batch is a <c>DELETE TOP (@batch)</c> so the
/// transaction log and lock footprint stay bounded regardless of backlog. Re-running
/// the purge is a no-op once every eligible row is gone (the loop exits when a batch
/// deletes zero rows), so a crash mid-loop is recoverable by simply running again.
/// the purge resumes remaining backlog after reaching <paramref name="maxRows"/>;
/// an empty batch ends it sooner. A crash mid-loop is recoverable by running again.
/// </para>
/// </remarks>
/// <param name="channel">Canonical channel name (the <c>Category</c> column value, e.g. <c>ApiOutbound</c>).</param>
@@ -183,13 +183,14 @@ public interface IAuditLogRepository
/// by the ~30s ADO.NET default (arch-review 04, S2).
/// </param>
/// <param name="ct">Cancellation token.</param>
/// <param name="maxRows">Maximum total rows deleted in this call; remaining backlog is retained for a later tick.</param>
/// <returns>A task that resolves to the total number of rows deleted across all batches.</returns>
Task<long> PurgeChannelOlderThanAsync(
string channel,
DateTime threshold,
int batchSize,
TimeSpan? commandTimeout = null,
CancellationToken ct = default);
CancellationToken ct = default, int maxRows = 50_000);
/// <summary>
/// Returns the set of <c>pf_AuditLog_Month</c> partition lower-bound
@@ -250,6 +250,10 @@ public interface ITemplateEngineRepository
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>A task that resolves to a read-only list of instances for the specified site.</returns>
Task<IReadOnlyList<Instance>> GetInstancesBySiteIdAsync(int siteId, CancellationToken cancellationToken = default);
/// <summary>Reads one bounded, untracked page of topology instance rows for a site/area.</summary>
Task<IReadOnlyList<Instance>> GetTopologyInstancesAsync(int siteId, int? areaId, int skip, int take, CancellationToken cancellationToken = default);
/// <summary>Retrieves an instance by unique name.</summary>
/// <param name="uniqueName">The unique instance name.</param>
/// <param name="cancellationToken">Cancellation token.</param>
@@ -17,11 +17,12 @@ public interface IPullTraceEventsClient
/// <param name="sinceUtc">Lower bound on <c>OccurredAtUtc</c> (the run's start is the natural choice).</param>
/// <param name="maxEvents">Row cap (the site caps at 2000).</param>
/// <param name="ct">Cancellation token.</param>
/// <param name="afterEventId">Optional exclusive EventId at sinceUtc for stable keyset paging.</param>
/// <returns>The matching rows ordered <c>(OccurredAtUtc, EventId)</c>; empty on fault or unknown site.</returns>
Task<IReadOnlyList<AuditEvent>> PullAsync(
string siteId,
IReadOnlyCollection<Guid> executionIds,
DateTime sinceUtc,
int maxEvents,
CancellationToken ct);
CancellationToken ct, string? afterEventId = null);
}
@@ -232,5 +232,5 @@ public interface ISiteAuditQueue
/// forwarding state — moves no cursor and flips nothing. At most 64 ids per call.
/// </summary>
Task<IReadOnlyList<AuditEvent>> ReadByExecutionIdsAsync(
IReadOnlyCollection<Guid> executionIds, DateTime sinceUtc, int limit, CancellationToken ct = default);
IReadOnlyCollection<Guid> executionIds, DateTime sinceUtc, int limit, CancellationToken ct = default, string? afterEventId = null);
}
@@ -12,13 +12,10 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
/// <param name="Quality">Quality of the new value (<c>Good</c>/<c>Bad</c>/<c>Uncertain</c>).</param>
/// <param name="Timestamp">UTC timestamp of the change.</param>
/// <param name="SourceExecutionId">
/// Audit Log #23 (ParentExecutionId tag-cascade): the <c>ExecutionId</c> of the
/// script execution / inbound API request whose write produced this change, or
/// <c>null</c> when the change did not originate from an audited execution —
/// which is the case for every value that arrives from the Data Connection
/// Layer (external device data), for deployment-time seeding, and for the
/// confirmed value of a data-sourced write (the device echo arrives on the
/// subscription long after the writing execution ended).
/// Execution associated with this change. Static writes carry an observed writer id;
/// armed device updates carry only a candidate writer id (see
/// <paramref name="SourceExecutionInferred"/>), since subscriptions carry no write-request id.
/// Other device updates and deployment seeding carry no id.
///
/// <para>
/// The Alarm Actor records this id as the <c>ParentExecutionId</c> of the
@@ -34,6 +31,7 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
/// </para>
/// </param>
/// <param name="Trace">Flow Debugger trace flag — see design 2026-09-05 §5.2. Additive; defaults to untraced.</param>
/// <param name="SourceExecutionInferred">True when the source is inferred from the first device update within the trace-arm window, regardless of value equality, clamping, or coercion; it is not evidence of a write echo.</param>
public record AttributeValueChanged(
string InstanceUniqueName,
string AttributePath,
@@ -42,4 +40,5 @@ public record AttributeValueChanged(
string Quality,
DateTimeOffset Timestamp,
Guid? SourceExecutionId = null,
bool Trace = false) : ISiteStreamEvent;
bool Trace = false,
bool SourceExecutionInferred = false) : ISiteStreamEvent;
@@ -66,16 +66,17 @@ public static class FlowGraphBuilder
// D2: a ScriptLog row is a line, never a node. Partition BEFORE the input-order index below, so
// `seq` indexes only real step rows and a line can never shift an execution's display position.
var lines = new List<FlowLogLine>();
var stepEvents = new List<AuditEvent>(events.Count);
var stepEvents = new List<(AuditEvent Event, AuditDetails Details)>(events.Count);
foreach (var e in events)
{
if (FlowStepProjector.IsLogLine(e))
lines.Add(FlowStepProjector.ToLogLine(e));
var details = AuditDetailsCodec.Deserialize(e.DetailsJson);
if (FlowStepProjector.IsLogLine(details))
lines.Add(FlowStepProjector.ToLogLine(e, details));
else
stepEvents.Add(e);
stepEvents.Add((e, details));
}
events = stepEvents;
events = stepEvents.Select(e => e.Event).ToList();
if (events.Count == 0)
return new FlowGraph(Array.Empty<FlowStep>(), Array.Empty<FlowEdge>(), Predicted: false, Truncated: truncated,
@@ -105,7 +106,7 @@ public static class FlowGraphBuilder
return index;
}
var folded = Fold(events).ToList();
var folded = Fold(stepEvents).ToList();
var firstSeqByExecution = folded
.GroupBy(s => s.ExecutionId ?? Guid.Empty)
.ToDictionary(g => g.Key, g => g.Min(InputIndex));
@@ -205,7 +206,7 @@ public static class FlowGraphBuilder
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));
edges.Add(new FlowEdge(spawner.EventId, firstChained.EventId, CrossesBoundary(spawner, firstChained), External: spawn && first.Extra?.TriggerSource == "inferred-device-update", Spawn: spawn));
}
// 3. Satisfied wait ⇐ preceding write in the same execution (inferred device echo).
@@ -219,6 +220,7 @@ public static class FlowGraphBuilder
for (var j = i - 1; j >= 0; j--)
{
if (chain[j].Kind != FlowStepKind.AttributeWrite) continue;
edges.RemoveAll(e => e.FromEventId == chain[j].EventId && e.ToEventId == wait.EventId);
edges.Add(new FlowEdge(chain[j].EventId, wait.EventId, CrossesBoundary: false, External: true));
break;
}
@@ -376,14 +378,14 @@ public static class FlowGraphBuilder
/// Projects every event and folds the outbound rows of one operation (same kind + <c>CorrelationId</c>) into one step
/// carrying the whole lifecycle. Non-outbound rows and null-correlation outbound rows pass through one-to-one.
/// </summary>
private static IEnumerable<FlowStep> Fold(IReadOnlyList<AuditEvent> events)
private static IEnumerable<FlowStep> Fold(IReadOnlyList<(AuditEvent Event, AuditDetails Details)> events)
{
var groups = new Dictionary<(FlowStepKind Kind, Guid Correlation), List<(FlowStep Step, AuditEvent Evt)>>();
var groups = new Dictionary<(FlowStepKind Kind, Guid Correlation), List<(FlowStep Step, FlowLifecycleRow Row)>>();
var order = new List<object>(events.Count); // FlowStep, or the group key — preserves first-seen order
foreach (var evt in events)
foreach (var (evt, details) in events)
{
var step = FlowStepProjector.Project(evt);
var step = FlowStepProjector.Project(evt, details);
if (!FlowStepProjector.IsOutbound(step.Kind))
{
order.Add(step);
@@ -392,7 +394,7 @@ public static class FlowGraphBuilder
if (evt.CorrelationId is not { } correlation)
{
order.Add(step with { Lifecycle = [FlowStepProjector.ToLifecycleRow(evt)] });
order.Add(step with { Lifecycle = [FlowStepProjector.ToLifecycleRow(evt, details)] });
continue;
}
@@ -403,7 +405,7 @@ public static class FlowGraphBuilder
groups[key] = group;
order.Add(key);
}
group.Add((step, evt));
group.Add((step, FlowStepProjector.ToLifecycleRow(evt, details)));
}
foreach (var item in order)
@@ -416,7 +418,6 @@ public static class FlowGraphBuilder
var key = ((FlowStepKind, Guid))item;
var rows = groups[key]
.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)
@@ -435,6 +436,7 @@ public static class FlowGraphBuilder
Subtitle = latest.Subtitle,
OutputJson = latest.OutputJson,
ErrorMessage = latest.ErrorMessage,
ErrorDetail = latest.ErrorDetail,
DurationMs = rows.Count == 1 && !terminal ? null : duration,
Lifecycle = lifecycle,
};
@@ -65,6 +65,7 @@ public enum FlowStepOutcome
/// <param name="Node">Node that wrote it.</param>
/// <param name="ErrorMessage">Error summary on a failed attempt.</param>
/// <param name="DurationMs">The row's own duration, when it has one.</param>
/// <param name="ErrorDetail">Exception detail from this lifecycle row.</param>
public sealed record FlowLifecycleRow(
Guid EventId,
string Kind,
@@ -72,7 +73,8 @@ public sealed record FlowLifecycleRow(
DateTimeOffset OccurredAtUtc,
string? Node,
string? ErrorMessage,
int? DurationMs);
int? DurationMs,
string? ErrorDetail = null);
/// <summary>
/// One log line folded onto a step (<see cref="FlowStep.Logs"/>; design 2026-09-09 D1/D8) — the
@@ -17,7 +17,11 @@ public static class FlowStepProjector
public static FlowStep Project(AuditEvent evt)
{
ArgumentNullException.ThrowIfNull(evt);
var d = AuditDetailsCodec.Deserialize(evt.DetailsJson);
return Project(evt, AuditDetailsCodec.Deserialize(evt.DetailsJson));
}
internal static FlowStep Project(AuditEvent evt, AuditDetails d)
{
var extra = ScriptTraceExtra.TryParse(d.Extra);
var kind = ResolveKind(d.Channel, d.Kind);
var lane = ResolveLane(kind, d.SourceSiteId);
@@ -52,6 +56,11 @@ public static class FlowStepProjector
{
ArgumentNullException.ThrowIfNull(evt);
var d = AuditDetailsCodec.Deserialize(evt.DetailsJson);
return IsLogLine(d);
}
internal static bool IsLogLine(AuditDetails d)
{
return string.Equals(d.Channel, nameof(AuditChannel.ScriptTrace), StringComparison.Ordinal)
&& string.Equals(d.Kind, nameof(AuditKind.ScriptLog), StringComparison.Ordinal);
}
@@ -60,7 +69,11 @@ public static class FlowStepProjector
public static FlowLogLine ToLogLine(AuditEvent evt)
{
ArgumentNullException.ThrowIfNull(evt);
var d = AuditDetailsCodec.Deserialize(evt.DetailsJson);
return ToLogLine(evt, AuditDetailsCodec.Deserialize(evt.DetailsJson));
}
internal static FlowLogLine ToLogLine(AuditEvent evt, AuditDetails d)
{
var x = ScriptTraceExtra.TryParse(d.Extra);
return new FlowLogLine(
EventId: evt.EventId,
@@ -79,7 +92,11 @@ public static class FlowStepProjector
public static FlowLifecycleRow ToLifecycleRow(AuditEvent evt)
{
ArgumentNullException.ThrowIfNull(evt);
var d = AuditDetailsCodec.Deserialize(evt.DetailsJson);
return ToLifecycleRow(evt, AuditDetailsCodec.Deserialize(evt.DetailsJson));
}
internal static FlowLifecycleRow ToLifecycleRow(AuditEvent evt, AuditDetails d)
{
return new FlowLifecycleRow(
EventId: evt.EventId,
Kind: d.Kind ?? string.Empty,
@@ -87,7 +104,8 @@ public static class FlowStepProjector
OccurredAtUtc: evt.OccurredAtUtc,
Node: evt.SourceNode,
ErrorMessage: d.ErrorMessage,
DurationMs: d.DurationMs);
DurationMs: d.DurationMs,
ErrorDetail: d.ErrorDetail);
}
/// <summary>Maps a channel/kind string pair onto a step kind.</summary>
@@ -22,7 +22,7 @@ public sealed record ScriptTraceExtra
/// <summary>Encoded value that fired a value-change run.</summary>
public string? TriggerValue { get; init; }
/// <summary>Where the triggering change came from: <c>value-change</c> (the actor cannot tell a static write from a device echo).</summary>
/// <summary>Where the triggering change came from: <c>value-change</c> for an observed write, or <c>inferred-device-update</c> for the first device publish in a trace-arm window (not a confirmed echo).</summary>
public string? TriggerSource { get; init; }
/// <summary>On a <c>RouteCall</c> row: the routed site run's <c>ExecutionId</c>.</summary>
@@ -554,5 +554,5 @@ public record RegisterNotificationOutbox(IActorRef OutboxProxy);
// NOTE: there is deliberately no RegisterAuditIngest counterpart. The audit-log-ingest
// singleton proxy goes STRAIGHT to the two gRPC servers that need it
// (CentralControlGrpcService.SetAuditIngestActor / SiteStreamGrpcServer.SetAuditIngestActor);
// (CentralControlGrpcService.SetAuditIngestActor);
// routing audit batches through this actor was a redundant hop and was removed.
@@ -6,6 +6,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication;
/// </summary>
public class CommunicationOptions
{
/// <summary>Maximum response-header wait before using the first stream event as the connected signal.</summary>
public TimeSpan ConnectedHeaderTimeout { get; set; } = TimeSpan.FromSeconds(10);
/// <summary>
/// Central control-plane gRPC endpoints (preferred first), e.g.
/// <c>["http://scadabridge-central-a:8083", "http://scadabridge-central-b:8083"]</c>. Dialled by
@@ -23,6 +23,9 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase<Communi
/// <inheritdoc />
protected override void Validate(ValidationBuilder builder, CommunicationOptions options)
{
builder.RequireThat(options.ConnectedHeaderTimeout > TimeSpan.Zero && options.ConnectedHeaderTimeout.TotalMilliseconds <= uint.MaxValue - 1,
"ScadaBridge:Communication:ConnectedHeaderTimeout must be positive and fit a timer.");
builder.RequireThat(options.DeploymentTimeout > TimeSpan.Zero,
$"ScadaBridge:Communication:DeploymentTimeout must be a positive duration (was {options.DeploymentTimeout}).");
@@ -39,17 +39,13 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
/// <see cref="SiteStreamGrpcServer.AuditIngestAskTimeout"/>, so the inner Ask expired at the same
/// instant as the outer one and could only add latency plus a
/// <see cref="Akka.Actor.Status.Failure"/> repackaging. The proxy now arrives through
/// <see cref="SetAuditIngestActor"/>, exactly as <see cref="SiteStreamGrpcServer"/> has always
/// taken it, and the relay handlers were deleted.
/// <see cref="SetAuditIngestActor"/>; the redundant actor relay handlers were deleted.
/// </para>
/// <para>
/// <b>Fault semantics deliberately differ from <see cref="SiteStreamGrpcServer"/>'s ingest
/// RPCs.</b> That server answers a failed audit ingest with an EMPTY <c>IngestAck</c>; this one
/// fails the call with a non-OK status. Both leave the site's rows <c>Pending</c> for the next
/// drain, so the outcome is the same — but this service replaced the ClusterClient path, whose
/// documented behaviour was to propagate the fault, and preserving that keeps a lost batch
/// visible as a failure rather than as a successful call that acked nothing. A proxy that has
/// not been handed over yet is NOT a fault and still answers with an empty ack.
/// <b>Fault semantics.</b> Failed central ingest returns a non-OK status, leaving the site's
/// rows Pending and making a lost batch visible to the next drain. A proxy not handed over
/// yet returns an empty ack. The legacy SiteStreamService ingest methods are retired and
/// always return Unimplemented; they are not an alternate ingestion path.
/// </para>
/// <para>
/// <b>Status mapping, and why it is not uniform.</b> A site transport may safely re-send a call
@@ -126,9 +122,7 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon
/// <summary>
/// Hands the central <c>AuditLogIngestActor</c> singleton proxy to the service so the two
/// ingest RPCs can Ask it DIRECTLY. Mirrors
/// <see cref="SiteStreamGrpcServer.SetAuditIngestActor"/> — the site-stream server has always
/// dispatched straight to the singleton, and this service now matches it.
/// ingest RPCs can Ask it directly. SiteStreamService retains only Unimplemented wire stubs.
/// </summary>
/// <remarks>
/// Removing the <c>CentralCommunicationActor</c> relay removes an actor hop, a second Ask,
@@ -52,9 +52,7 @@ public sealed class SitePairChannelProvider : IDisposable
/// <summary>
/// How long a replaced/removed channel is PARKED before it is disposed (arch-review
/// 2026-08-19 P3-2). Sized to exceed the longest command deadline
/// (<c>CommunicationOptions.DeploymentTimeout</c> and friends, minutes at most), so no call
/// that was already in flight when the channel was replaced can still be using it.
/// 2026-08-19 P3-2). Active calls retain the channel beyond this minimum until completion.
/// </summary>
internal static readonly TimeSpan ChannelDisposalGrace = TimeSpan.FromMinutes(5);
@@ -68,15 +66,16 @@ public sealed class SitePairChannelProvider : IDisposable
/// <summary>
/// Channels replaced by <see cref="UpdateSite"/> or dropped by <see cref="RemoveSite"/>,
/// each with the instant it becomes safe to dispose. FIFO order is disposal order because
/// the grace window is constant.
/// each with the earliest idle disposal instant. Active-call retention can extend that instant.
/// </summary>
private readonly ConcurrentQueue<(GrpcChannel Channel, DateTime DisposeAfterUtc)> _graveyard = new();
private readonly ConcurrentDictionary<GrpcChannel, int> _activeCalls = new();
private readonly CancellationTokenSource _shutdown = new();
private Task? _failbackLoop;
private readonly object _loopGate = new();
private bool _disposed;
private volatile bool _disposed;
private readonly object _lifecycleGate = new();
/// <summary>Creates the provider.</summary>
/// <param name="pskProvider">Resolves each site's preshared key for channel credentials.</param>
@@ -120,38 +119,43 @@ public sealed class SitePairChannelProvider : IDisposable
/// <param name="nodeBEndpoint">NodeB gRPC base address, or null when unconfigured.</param>
public void UpdateSite(string siteId, string? nodeAEndpoint, string? nodeBEndpoint)
{
ArgumentException.ThrowIfNullOrWhiteSpace(siteId);
SweepGraveyard();
var pair = _sites.GetOrAdd(siteId, id => new SitePair(id));
lock (pair.Gate)
lock (_lifecycleGate)
{
var changedA = !string.Equals(pair.NodeAEndpoint, nodeAEndpoint, StringComparison.Ordinal);
var changedB = !string.Equals(pair.NodeBEndpoint, nodeBEndpoint, StringComparison.Ordinal);
if (!changedA && !changedB)
ObjectDisposedException.ThrowIf(_disposed, this);
ArgumentException.ThrowIfNullOrWhiteSpace(siteId);
SweepGraveyard();
var pair = _sites.GetOrAdd(siteId, id => new SitePair(id));
lock (pair.Gate)
{
return;
var changedA = !string.Equals(pair.NodeAEndpoint, nodeAEndpoint, StringComparison.Ordinal);
var changedB = !string.Equals(pair.NodeBEndpoint, nodeBEndpoint, StringComparison.Ordinal);
if (!changedA && !changedB)
{
return;
}
if (changedA)
{
Park(pair.ChannelA);
pair.ChannelA = string.IsNullOrWhiteSpace(nodeAEndpoint) ? null : BuildChannel(nodeAEndpoint, siteId);
pair.NodeAEndpoint = nodeAEndpoint;
}
if (changedB)
{
Park(pair.ChannelB);
pair.ChannelB = string.IsNullOrWhiteSpace(nodeBEndpoint) ? null : BuildChannel(nodeBEndpoint, siteId);
pair.NodeBEndpoint = nodeBEndpoint;
}
// Addresses changed — return to the preferred node and reset failback backoff.
pair.CurrentIsA = pair.ChannelA is not null;
pair.ResetFailback();
_logger.LogInformation(
"gRPC channel pair for site {SiteId} refreshed (nodeA={HasA}, nodeB={HasB})",
siteId, pair.ChannelA is not null, pair.ChannelB is not null);
}
if (changedA)
{
Park(pair.ChannelA);
pair.ChannelA = string.IsNullOrWhiteSpace(nodeAEndpoint) ? null : BuildChannel(nodeAEndpoint, siteId);
pair.NodeAEndpoint = nodeAEndpoint;
}
if (changedB)
{
Park(pair.ChannelB);
pair.ChannelB = string.IsNullOrWhiteSpace(nodeBEndpoint) ? null : BuildChannel(nodeBEndpoint, siteId);
pair.NodeBEndpoint = nodeBEndpoint;
}
// Addresses changed — return to the preferred node and reset failback backoff.
pair.CurrentIsA = pair.ChannelA is not null;
pair.ResetFailback();
_logger.LogInformation(
"gRPC channel pair for site {SiteId} refreshed (nodeA={HasA}, nodeB={HasB})",
siteId, pair.ChannelA is not null, pair.ChannelB is not null);
}
}
@@ -162,24 +166,29 @@ public sealed class SitePairChannelProvider : IDisposable
/// <param name="siteId">Site identifier to remove.</param>
public void RemoveSite(string siteId)
{
if (string.IsNullOrWhiteSpace(siteId))
lock (_lifecycleGate)
{
return;
}
SweepGraveyard();
if (_sites.TryRemove(siteId, out var pair))
{
lock (pair.Gate)
ObjectDisposedException.ThrowIf(_disposed, this);
if (string.IsNullOrWhiteSpace(siteId))
{
Park(pair.ChannelA);
Park(pair.ChannelB);
pair.ChannelA = null;
pair.ChannelB = null;
return;
}
_pskProvider.Invalidate(siteId);
_logger.LogInformation("gRPC channel pair for site {SiteId} removed", siteId);
SweepGraveyard();
if (_sites.TryRemove(siteId, out var pair))
{
lock (pair.Gate)
{
Park(pair.ChannelA);
Park(pair.ChannelB);
pair.ChannelA = null;
pair.ChannelB = null;
}
_pskProvider.Invalidate(siteId);
_logger.LogInformation("gRPC channel pair for site {SiteId} removed", siteId);
}
}
}
@@ -219,6 +228,8 @@ public sealed class SitePairChannelProvider : IDisposable
// Recompute which node `primary` actually is (the current side may be null).
primaryIsA = ReferenceEquals(primary, pair.ChannelA);
secondary = primaryIsA ? pair.ChannelB : pair.ChannelA;
_activeCalls.AddOrUpdate(primary, 1, (_, count) => count + 1);
if (secondary is not null) _activeCalls.AddOrUpdate(secondary, 1, (_, count) => count + 1);
}
try
@@ -242,6 +253,12 @@ public sealed class SitePairChannelProvider : IDisposable
throw GrpcRetrySafety.OutcomeUnknown(retryError);
}
}
finally
{
_activeCalls.AddOrUpdate(primary, 0, (_, count) => count - 1);
if (secondary is not null) _activeCalls.AddOrUpdate(secondary, 0, (_, count) => count - 1);
if (_disposed) SweepGraveyard(force: true);
}
}
/// <summary>
@@ -360,13 +377,8 @@ public sealed class SitePairChannelProvider : IDisposable
/// and the command simply failed.
/// </para>
/// <para>
/// <b>Why a timed park rather than reference counting.</b> A refcount would have to be taken
/// and released around every <c>ExecuteAsync</c> call — bookkeeping on the hot path, plus a
/// quiescence handshake — to buy nothing a fixed window does not already give: the window
/// only has to outlast the longest command deadline. It also matches the precedent set by
/// <see cref="SiteStreamGrpcClientFactory"/>, which likewise keeps a superseded endpoint's
/// channel alive rather than disposing it under a concurrent session. Cost is a handful of
/// idle HTTP/2 channels for at most <see cref="ChannelDisposalGrace"/>.
/// Active calls retain both selected channels until completion, including the retry channel.
/// The grace is only an idle-channel minimum; dynamic WaitForAttribute deadlines can exceed it.
/// </para>
/// </summary>
/// <param name="channel">The channel to park; null is ignored.</param>
@@ -383,25 +395,27 @@ public sealed class SitePairChannelProvider : IDisposable
/// path (<see cref="UpdateSite"/>/<see cref="RemoveSite"/>) and from the failback loop, so a
/// park always drains even on a provider that never runs a call again.
/// </summary>
/// <param name="force">True to dispose every parked channel regardless of its window (shutdown, and tests).</param>
/// <param name="force">True to ignore idle grace (shutdown and tests); active calls always retain their channels.</param>
internal void SweepGraveyard(bool force = false)
{
var now = DateTime.UtcNow;
while (_graveyard.TryPeek(out var parked) && (force || now >= parked.DisposeAfterUtc))
lock (_lifecycleGate)
{
if (!_graveyard.TryDequeue(out parked))
var now = DateTime.UtcNow;
var count = _graveyard.Count;
// Visit each parked channel once: an occupied channel must not starve later entries.
for (var i = 0; i < count && _graveyard.TryDequeue(out var parked); i++)
{
break;
if ((!force && now < parked.DisposeAfterUtc)
|| _activeCalls.GetValueOrDefault(parked.Channel) > 0)
{
_graveyard.Enqueue(parked);
continue;
}
_activeCalls.TryRemove(parked.Channel, out _);
try { parked.Channel.Dispose(); }
catch (Exception ex) { _logger.LogDebug(ex, "Disposing a parked gRPC channel faulted"); }
}
try
{
parked.Channel.Dispose();
}
catch (Exception ex)
{
_logger.LogDebug(ex, "Disposing a parked gRPC channel faulted");
}
}
}
@@ -461,39 +475,29 @@ public sealed class SitePairChannelProvider : IDisposable
/// <inheritdoc />
public void Dispose()
{
if (_disposed)
lock (_lifecycleGate)
{
return;
}
_disposed = true;
_shutdown.Cancel();
try
{
_failbackLoop?.Wait(TimeSpan.FromSeconds(2));
}
catch
{
// Best-effort loop drain on shutdown.
}
_shutdown.Dispose();
foreach (var pair in _sites.Values)
{
lock (pair.Gate)
if (_disposed) return;
_disposed = true;
foreach (var pair in _sites.Values)
{
pair.ChannelA?.Dispose();
pair.ChannelB?.Dispose();
lock (pair.Gate)
{
Park(pair.ChannelA);
Park(pair.ChannelB);
pair.ChannelA = null;
pair.ChannelB = null;
}
}
_sites.Clear();
}
_sites.Clear();
// Process shutdown — nothing can still be in flight, so drain the graveyard regardless
// of the remaining grace on each entry.
_shutdown.Cancel();
try { _failbackLoop?.Wait(TimeSpan.FromSeconds(2)); }
catch { /* shutdown is best effort; active calls retain their channels */ }
_shutdown.Dispose();
SweepGraveyard(force: true);
}
/// <summary>Per-site mutable channel state, guarded by <see cref="Gate"/>.</summary>
private sealed class SitePair(string siteId)
{
public string SiteId { get; } = siteId;
@@ -82,6 +82,7 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
ISitePskProvider? pskProvider,
string? siteIdentifier)
{
ConnectedHeaderTimeout = options.ConnectedHeaderTimeout;
Endpoint = endpoint;
KeepAlivePingDelay = options.GrpcKeepAlivePingDelay;
KeepAlivePingTimeout = options.GrpcKeepAlivePingTimeout;
@@ -122,7 +123,12 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
/// Unsubscribe and Dispose behavior without needing a real endpoint.
/// </summary>
/// <returns>A <see cref="SiteStreamGrpcClient"/> with no channel or client, for testing only.</returns>
internal static SiteStreamGrpcClient CreateForTesting() => new();
internal static SiteStreamGrpcClient CreateForTesting(CommunicationOptions? options = null) => new(options ?? new CommunicationOptions());
private SiteStreamGrpcClient(CommunicationOptions options)
{
ConnectedHeaderTimeout = options.ConnectedHeaderTimeout;
}
/// <summary>
/// Registers a CancellationTokenSource for a correlation ID. Test-only.
@@ -457,7 +463,7 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
/// with its first message would otherwise hold the connected signal — and with it the
/// aggregator's re-seed — for as long as the site happens to be quiet.
/// </summary>
internal static TimeSpan ConnectedHeaderTimeout { get; set; } = TimeSpan.FromSeconds(10);
internal TimeSpan ConnectedHeaderTimeout { get; } = TimeSpan.FromSeconds(10);
/// <summary>
/// Awaits the call's response headers, bounded by <see cref="ConnectedHeaderTimeout"/>.
@@ -475,7 +481,7 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
/// timeout was added for is covered by the reconcile backstop.
/// </para>
/// </summary>
private static async Task<bool> AwaitHeadersAsync(
private async Task<bool> AwaitHeadersAsync(
AsyncServerStreamingCall<SiteStreamEvent> call, CancellationToken ct)
{
var headers = call.ResponseHeadersAsync;
@@ -38,11 +38,6 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
// true, never reset (the server is single-lifetime per host).
private volatile bool _shuttingDown;
private long _actorCounter;
// Central-side ingest actor proxy. Set by the host
// after the cluster singleton starts. When null the
// IngestAuditEvents RPC replies with an empty IngestAck so sites can
// safely retry — wiring-incomplete is treated as transient, never fatal.
private IActorRef? _auditIngestActor;
// Ask timeout is 30 s. The ingest actor's repo
// calls are sub-100 ms in steady state; a generous timeout absorbs a slow
// MSSQL connection without surfacing as a gRPC failure on a healthy site.
@@ -153,26 +148,12 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
_ready = true;
}
/// <summary>
/// Hands the central-side <c>AuditLogIngestActor</c> proxy to the gRPC
/// server so the <see cref="IngestAuditEvents"/> RPC can route incoming
/// site batches. Mirrors the way
/// <c>CommunicationService.SetNotificationOutbox</c> takes the Notification
/// Outbox singleton proxy. The actor is supplied after the cluster
/// singleton starts.
/// </summary>
/// <param name="proxy">The audit log ingest actor proxy.</param>
public void SetAuditIngestActor(IActorRef proxy)
{
_auditIngestActor = proxy;
}
/// <summary>
/// Hands the site-local <see cref="ISiteAuditQueue"/> (the same
/// <c>SqliteAuditWriter</c> singleton that backs <see cref="IAuditWriter"/>
/// on the script thread) to the gRPC server so the
/// <see cref="PullAuditEvents"/> RPC can serve central's reconciliation
/// pulls. Mirrors <see cref="SetAuditIngestActor"/>: wired post-construction
/// pulls. Wired post-construction
/// because the queue and the gRPC server are both DI singletons brought up
/// in independent orders on site startup.
/// </summary>
@@ -535,111 +516,18 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
}
}
/// <inheritdoc />
public override async Task<IngestAck> IngestAuditEvents(
AuditEventBatch request,
ServerCallContext context)
{
// Empty batch is a no-op; reply immediately so the client moves on.
if (request.Events.Count == 0)
{
return new IngestAck();
}
/// <summary>Retained wire method; ingest is served only by CentralControlService.</summary>
public override Task<IngestAck> IngestAuditEvents(AuditEventBatch request, ServerCallContext context)
=> Task.FromException<IngestAck>(new RpcException(new GrpcStatus(
StatusCode.Unimplemented, "Use CentralControlService.IngestAuditEvents on central.")));
var actor = _auditIngestActor;
if (actor is null)
{
// Wiring incomplete (host startup race). Sites treat an empty
// ack as "nothing was acked, leave rows Pending, retry next
// drain" — exactly the right behaviour during host bring-up.
_logger.LogWarning(
"IngestAuditEvents received {Count} events before SetAuditIngestActor was called; returning empty ack.",
request.Events.Count);
return new IngestAck();
}
/// <summary>Retained wire method; combined ingest is served only by CentralControlService.</summary>
public override Task<IngestAck> IngestCachedTelemetry(CachedTelemetryBatch request, ServerCallContext context)
=> Task.FromException<IngestAck>(new RpcException(new GrpcStatus(
StatusCode.Unimplemented, "Use CentralControlService.IngestCachedTelemetry on central.")));
var entities = new List<AuditEvent>(request.Events.Count);
foreach (var dto in request.Events)
{
entities.Add(AuditEventDtoMapper.FromDto(dto));
}
var cmd = new IngestAuditEventsCommand(entities);
IngestAuditEventsReply reply;
try
{
reply = await actor.Ask<IngestAuditEventsReply>(
cmd, AuditIngestAskTimeout, context.CancellationToken);
}
catch (Exception ex)
{
// Audit ingest is best-effort; failing this RPC at the gRPC layer
// would surface as a transport error and force the site to retry
// (which it would do anyway). Logging + an empty ack keeps the
// semantics consistent with the "wiring incomplete" path above.
_logger.LogError(ex,
"AuditLogIngestActor Ask failed for batch of {Count} events; returning empty ack.",
request.Events.Count);
return new IngestAck();
}
var ack = new IngestAck();
foreach (var id in reply.AcceptedEventIds)
{
ack.AcceptedEventIds.Add(id.ToString());
}
return ack;
}
/// <inheritdoc />
public override async Task<IngestAck> IngestCachedTelemetry(
CachedTelemetryBatch request,
ServerCallContext context)
{
if (request.Packets.Count == 0)
{
return new IngestAck();
}
var actor = _auditIngestActor;
if (actor is null)
{
_logger.LogWarning(
"IngestCachedTelemetry received {Count} packets before SetAuditIngestActor was called; returning empty ack.",
request.Packets.Count);
return new IngestAck();
}
var entries = new List<CachedTelemetryEntry>(request.Packets.Count);
foreach (var packet in request.Packets)
{
var auditEvent = AuditEventDtoMapper.FromDto(packet.AuditEvent);
var siteCall = SiteCallDtoMapper.FromDto(packet.Operational);
entries.Add(new CachedTelemetryEntry(auditEvent, siteCall));
}
var cmd = new IngestCachedTelemetryCommand(entries);
IngestCachedTelemetryReply reply;
try
{
reply = await actor.Ask<IngestCachedTelemetryReply>(
cmd, AuditIngestAskTimeout, context.CancellationToken);
}
catch (Exception ex)
{
_logger.LogError(ex,
"AuditLogIngestActor Ask failed for combined telemetry batch of {Count} packets; returning empty ack.",
request.Packets.Count);
return new IngestAck();
}
var ack = new IngestAck();
foreach (var id in reply.AcceptedEventIds)
{
ack.AcceptedEventIds.Add(id.ToString());
}
return ack;
}
/// <summary>Maximum rows requested by one reconciliation pull, before the wire-byte budget.</summary>
public const int MaxReconciliationRowsPerPull = 2000;
/// <summary>Flow Debugger PullTraceEvents: server-side row cap (design 2026-09-05 §5.4).</summary>
public const int MaxTraceEventsPerPull = 2000;
@@ -694,7 +582,8 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
IReadOnlyList<AuditEvent> events;
try
{
events = await queue.ReadByExecutionIdsAsync(ids, since, request.MaxEvents, context.CancellationToken);
events = await queue.ReadByExecutionIdsAsync(ids, since, request.MaxEvents, context.CancellationToken,
string.IsNullOrEmpty(request.AfterId) ? null : request.AfterId);
}
catch (Exception ex)
{
@@ -725,13 +614,13 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
return new PullAuditEventsResponse();
}
if (request.BatchSize <= 0)
if (request.BatchSize <= 0 || request.BatchSize > MaxReconciliationRowsPerPull)
{
// Mirrors the SubscribeInstance guard: reject malformed requests
// cleanly with InvalidArgument so the caller doesn't see a generic
// RpcException from the underlying SQLite parameter validation.
throw new RpcException(new GrpcStatus(
StatusCode.InvalidArgument, "batch_size must be > 0"));
StatusCode.InvalidArgument, $"batch_size must be 1..{MaxReconciliationRowsPerPull}"));
}
// sinceUtc defaults to DateTime.MinValue when the wrapper is absent —
@@ -818,13 +707,13 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
return new PullSiteCallsResponse();
}
if (request.BatchSize <= 0)
if (request.BatchSize <= 0 || request.BatchSize > MaxReconciliationRowsPerPull)
{
// Mirrors PullAuditEvents: reject malformed requests cleanly with
// InvalidArgument so the caller doesn't see a generic RpcException
// from the underlying SQLite parameter validation.
throw new RpcException(new GrpcStatus(
StatusCode.InvalidArgument, "batch_size must be > 0"));
StatusCode.InvalidArgument, $"batch_size must be 1..{MaxReconciliationRowsPerPull}"));
}
// since_utc defaults to DateTime.MinValue when the wrapper is absent —
@@ -234,6 +234,7 @@ message PullAuditEventsResponse {
// OccurredAtUtc >= since_utc, ordered (OccurredAtUtc, EventId), at most max_events (server
// caps at 2000). Ids capped at 64 per call.
message PullTraceEventsRequest {
string after_id = 4; // Exclusive EventId at since_utc; empty preserves the inclusive legacy read.
repeated string execution_ids = 1; // "D" GUID form
google.protobuf.Timestamp since_utc = 2;
int32 max_events = 3;
@@ -90,38 +90,38 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
"KAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhIKCmJhdGNoX3NpemUY",
"AiABKAUSEAoIYWZ0ZXJfaWQYAyABKAkiXAoXUHVsbEF1ZGl0RXZlbnRzUmVz",
"cG9uc2USKQoGZXZlbnRzGAEgAygLMhkuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50",
"RHRvEhYKDm1vcmVfYXZhaWxhYmxlGAIgASgIInIKFlB1bGxUcmFjZUV2ZW50",
"c1JlcXVlc3QSFQoNZXhlY3V0aW9uX2lkcxgBIAMoCRItCglzaW5jZV91dGMY",
"AiABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhIKCm1heF9ldmVu",
"dHMYAyABKAUiRAoXUHVsbFRyYWNlRXZlbnRzUmVzcG9uc2USKQoGZXZlbnRz",
"GAEgAygLMhkuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50RHRvImsKFFB1bGxTaXRl",
"Q2FsbHNSZXF1ZXN0Ei0KCXNpbmNlX3V0YxgBIAEoCzIaLmdvb2dsZS5wcm90",
"b2J1Zi5UaW1lc3RhbXASEgoKYmF0Y2hfc2l6ZRgCIAEoBRIQCghhZnRlcl9p",
"ZBgDIAEoCSJpChVQdWxsU2l0ZUNhbGxzUmVzcG9uc2USOAoMb3BlcmF0aW9u",
"YWxzGAEgAygLMiIuc2l0ZXN0cmVhbS5TaXRlQ2FsbE9wZXJhdGlvbmFsRHRv",
"EhYKDm1vcmVfYXZhaWxhYmxlGAIgASgIKlwKB1F1YWxpdHkSFwoTUVVBTElU",
"WV9VTlNQRUNJRklFRBAAEhAKDFFVQUxJVFlfR09PRBABEhUKEVFVQUxJVFlf",
"VU5DRVJUQUlOEAISDwoLUVVBTElUWV9CQUQQAypdCg5BbGFybVN0YXRlRW51",
"bRIbChdBTEFSTV9TVEFURV9VTlNQRUNJRklFRBAAEhYKEkFMQVJNX1NUQVRF",
"X05PUk1BTBABEhYKEkFMQVJNX1NUQVRFX0FDVElWRRACKoUBCg5BbGFybUxl",
"dmVsRW51bRIUChBBTEFSTV9MRVZFTF9OT05FEAASEwoPQUxBUk1fTEVWRUxf",
"TE9XEAESFwoTQUxBUk1fTEVWRUxfTE9XX0xPVxACEhQKEEFMQVJNX0xFVkVM",
"X0hJR0gQAxIZChVBTEFSTV9MRVZFTF9ISUdIX0hJR0gQBDLiBAoRU2l0ZVN0",
"cmVhbVNlcnZpY2USVQoRU3Vic2NyaWJlSW5zdGFuY2USIS5zaXRlc3RyZWFt",
"Lkluc3RhbmNlU3RyZWFtUmVxdWVzdBobLnNpdGVzdHJlYW0uU2l0ZVN0cmVh",
"bUV2ZW50MAESTQoNU3Vic2NyaWJlU2l0ZRIdLnNpdGVzdHJlYW0uU2l0ZVN0",
"cmVhbVJlcXVlc3QaGy5zaXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVudDABEkcK",
"EUluZ2VzdEF1ZGl0RXZlbnRzEhsuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50QmF0",
"Y2gaFS5zaXRlc3RyZWFtLkluZ2VzdEFjaxJQChVJbmdlc3RDYWNoZWRUZWxl",
"bWV0cnkSIC5zaXRlc3RyZWFtLkNhY2hlZFRlbGVtZXRyeUJhdGNoGhUuc2l0",
"ZXN0cmVhbS5Jbmdlc3RBY2sSWgoPUHVsbEF1ZGl0RXZlbnRzEiIuc2l0ZXN0",
"cmVhbS5QdWxsQXVkaXRFdmVudHNSZXF1ZXN0GiMuc2l0ZXN0cmVhbS5QdWxs",
"QXVkaXRFdmVudHNSZXNwb25zZRJaCg9QdWxsVHJhY2VFdmVudHMSIi5zaXRl",
"c3RyZWFtLlB1bGxUcmFjZUV2ZW50c1JlcXVlc3QaIy5zaXRlc3RyZWFtLlB1",
"bGxUcmFjZUV2ZW50c1Jlc3BvbnNlElQKDVB1bGxTaXRlQ2FsbHMSIC5zaXRl",
"c3RyZWFtLlB1bGxTaXRlQ2FsbHNSZXF1ZXN0GiEuc2l0ZXN0cmVhbS5QdWxs",
"U2l0ZUNhbGxzUmVzcG9uc2VCK6oCKFpCLk1PTS5XVy5TY2FkYUJyaWRnZS5D",
"b21tdW5pY2F0aW9uLkdycGNiBnByb3RvMw=="));
"RHRvEhYKDm1vcmVfYXZhaWxhYmxlGAIgASgIIoQBChZQdWxsVHJhY2VFdmVu",
"dHNSZXF1ZXN0EhAKCGFmdGVyX2lkGAQgASgJEhUKDWV4ZWN1dGlvbl9pZHMY",
"ASADKAkSLQoJc2luY2VfdXRjGAIgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRp",
"bWVzdGFtcBISCgptYXhfZXZlbnRzGAMgASgFIkQKF1B1bGxUcmFjZUV2ZW50",
"c1Jlc3BvbnNlEikKBmV2ZW50cxgBIAMoCzIZLnNpdGVzdHJlYW0uQXVkaXRF",
"dmVudER0byJrChRQdWxsU2l0ZUNhbGxzUmVxdWVzdBItCglzaW5jZV91dGMY",
"ASABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhIKCmJhdGNoX3Np",
"emUYAiABKAUSEAoIYWZ0ZXJfaWQYAyABKAkiaQoVUHVsbFNpdGVDYWxsc1Jl",
"c3BvbnNlEjgKDG9wZXJhdGlvbmFscxgBIAMoCzIiLnNpdGVzdHJlYW0uU2l0",
"ZUNhbGxPcGVyYXRpb25hbER0bxIWCg5tb3JlX2F2YWlsYWJsZRgCIAEoCCpc",
"CgdRdWFsaXR5EhcKE1FVQUxJVFlfVU5TUEVDSUZJRUQQABIQCgxRVUFMSVRZ",
"X0dPT0QQARIVChFRVUFMSVRZX1VOQ0VSVEFJThACEg8KC1FVQUxJVFlfQkFE",
"EAMqXQoOQWxhcm1TdGF0ZUVudW0SGwoXQUxBUk1fU1RBVEVfVU5TUEVDSUZJ",
"RUQQABIWChJBTEFSTV9TVEFURV9OT1JNQUwQARIWChJBTEFSTV9TVEFURV9B",
"Q1RJVkUQAiqFAQoOQWxhcm1MZXZlbEVudW0SFAoQQUxBUk1fTEVWRUxfTk9O",
"RRAAEhMKD0FMQVJNX0xFVkVMX0xPVxABEhcKE0FMQVJNX0xFVkVMX0xPV19M",
"T1cQAhIUChBBTEFSTV9MRVZFTF9ISUdIEAMSGQoVQUxBUk1fTEVWRUxfSElH",
"SF9ISUdIEAQy4gQKEVNpdGVTdHJlYW1TZXJ2aWNlElUKEVN1YnNjcmliZUlu",
"c3RhbmNlEiEuc2l0ZXN0cmVhbS5JbnN0YW5jZVN0cmVhbVJlcXVlc3QaGy5z",
"aXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVudDABEk0KDVN1YnNjcmliZVNpdGUS",
"HS5zaXRlc3RyZWFtLlNpdGVTdHJlYW1SZXF1ZXN0Ghsuc2l0ZXN0cmVhbS5T",
"aXRlU3RyZWFtRXZlbnQwARJHChFJbmdlc3RBdWRpdEV2ZW50cxIbLnNpdGVz",
"dHJlYW0uQXVkaXRFdmVudEJhdGNoGhUuc2l0ZXN0cmVhbS5Jbmdlc3RBY2sS",
"UAoVSW5nZXN0Q2FjaGVkVGVsZW1ldHJ5EiAuc2l0ZXN0cmVhbS5DYWNoZWRU",
"ZWxlbWV0cnlCYXRjaBoVLnNpdGVzdHJlYW0uSW5nZXN0QWNrEloKD1B1bGxB",
"dWRpdEV2ZW50cxIiLnNpdGVzdHJlYW0uUHVsbEF1ZGl0RXZlbnRzUmVxdWVz",
"dBojLnNpdGVzdHJlYW0uUHVsbEF1ZGl0RXZlbnRzUmVzcG9uc2USWgoPUHVs",
"bFRyYWNlRXZlbnRzEiIuc2l0ZXN0cmVhbS5QdWxsVHJhY2VFdmVudHNSZXF1",
"ZXN0GiMuc2l0ZXN0cmVhbS5QdWxsVHJhY2VFdmVudHNSZXNwb25zZRJUCg1Q",
"dWxsU2l0ZUNhbGxzEiAuc2l0ZXN0cmVhbS5QdWxsU2l0ZUNhbGxzUmVxdWVz",
"dBohLnNpdGVzdHJlYW0uUHVsbFNpdGVDYWxsc1Jlc3BvbnNlQiuqAihaQi5N",
"T00uV1cuU2NhZGFCcmlkZ2UuQ29tbXVuaWNhdGlvbi5HcnBjYgZwcm90bzM="));
descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData,
new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.TimestampReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.WrappersReflection.Descriptor, },
new pbr::GeneratedClrTypeInfo(new[] {typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.Quality), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateEnum), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmLevelEnum), }, null, new pbr::GeneratedClrTypeInfo[] {
@@ -139,7 +139,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryBatch), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryBatch.Parser, new[]{ "Packets" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsRequest.Parser, new[]{ "SinceUtc", "BatchSize", "AfterId" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse.Parser, new[]{ "Events", "MoreAvailable" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest.Parser, new[]{ "ExecutionIds", "SinceUtc", "MaxEvents" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest.Parser, new[]{ "AfterId", "ExecutionIds", "SinceUtc", "MaxEvents" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsResponse), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsResponse.Parser, new[]{ "Events" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest.Parser, new[]{ "SinceUtc", "BatchSize", "AfterId" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse.Parser, new[]{ "Operationals", "MoreAvailable" }, null, null, null, null)
@@ -1418,8 +1418,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
private string sourceExecutionId_ = "";
/// <summary>
/// Flow Debugger (design 2026-09-05 §5.3/§5.4): the writing execution's id ("D" GUID
/// form; empty represents null) and the trace flag, so the debug stream can hand the
/// debugger the spawned run's parent. Additive; an older site never sets them.
/// form; empty represents null) and the trace flag. Additive; an older site never sets them.
/// RESERVED — populated by the site, consumed by nothing (arch-review F12): the value-change
/// entry polls the site's PullTraceEvents RPC instead of subscribing to the debug stream, so
/// these two fields are kept for a future debug-stream path. Do not reuse the field numbers.
/// </summary>
[global::System.Diagnostics.DebuggerNonUserCodeAttribute]
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
@@ -5942,6 +5944,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
[global::System.Diagnostics.DebuggerNonUserCodeAttribute]
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
public PullTraceEventsRequest(PullTraceEventsRequest other) : this() {
afterId_ = other.afterId_;
executionIds_ = other.executionIds_.Clone();
sinceUtc_ = other.sinceUtc_ != null ? other.sinceUtc_.Clone() : null;
maxEvents_ = other.maxEvents_;
@@ -5954,6 +5957,21 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
return new PullTraceEventsRequest(this);
}
/// <summary>Field number for the "after_id" field.</summary>
public const int AfterIdFieldNumber = 4;
private string afterId_ = "";
/// <summary>
/// Exclusive EventId at since_utc; empty preserves the inclusive legacy read.
/// </summary>
[global::System.Diagnostics.DebuggerNonUserCodeAttribute]
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
public string AfterId {
get { return afterId_; }
set {
afterId_ = pb::ProtoPreconditions.CheckNotNull(value, "value");
}
}
/// <summary>Field number for the "execution_ids" field.</summary>
public const int ExecutionIdsFieldNumber = 1;
private static readonly pb::FieldCodec<string> _repeated_executionIds_codec
@@ -6007,6 +6025,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
if (ReferenceEquals(other, this)) {
return true;
}
if (AfterId != other.AfterId) return false;
if(!executionIds_.Equals(other.executionIds_)) return false;
if (!object.Equals(SinceUtc, other.SinceUtc)) return false;
if (MaxEvents != other.MaxEvents) return false;
@@ -6017,6 +6036,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
public override int GetHashCode() {
int hash = 1;
if (AfterId.Length != 0) hash ^= AfterId.GetHashCode();
hash ^= executionIds_.GetHashCode();
if (sinceUtc_ != null) hash ^= SinceUtc.GetHashCode();
if (MaxEvents != 0) hash ^= MaxEvents.GetHashCode();
@@ -6047,6 +6067,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
output.WriteRawTag(24);
output.WriteInt32(MaxEvents);
}
if (AfterId.Length != 0) {
output.WriteRawTag(34);
output.WriteString(AfterId);
}
if (_unknownFields != null) {
_unknownFields.WriteTo(output);
}
@@ -6066,6 +6090,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
output.WriteRawTag(24);
output.WriteInt32(MaxEvents);
}
if (AfterId.Length != 0) {
output.WriteRawTag(34);
output.WriteString(AfterId);
}
if (_unknownFields != null) {
_unknownFields.WriteTo(ref output);
}
@@ -6076,6 +6104,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
public int CalculateSize() {
int size = 0;
if (AfterId.Length != 0) {
size += 1 + pb::CodedOutputStream.ComputeStringSize(AfterId);
}
size += executionIds_.CalculateSize(_repeated_executionIds_codec);
if (sinceUtc_ != null) {
size += 1 + pb::CodedOutputStream.ComputeMessageSize(SinceUtc);
@@ -6095,6 +6126,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
if (other == null) {
return;
}
if (other.AfterId.Length != 0) {
AfterId = other.AfterId;
}
executionIds_.Add(other.executionIds_);
if (other.sinceUtc_ != null) {
if (sinceUtc_ == null) {
@@ -6139,6 +6173,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
MaxEvents = input.ReadInt32();
break;
}
case 34: {
AfterId = input.ReadString();
break;
}
}
}
#endif
@@ -6173,6 +6211,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
MaxEvents = input.ReadInt32();
break;
}
case 34: {
AfterId = input.ReadString();
break;
}
}
}
}
@@ -649,7 +649,7 @@ VALUES
DateTime threshold,
int batchSize,
TimeSpan? commandTimeout = null,
CancellationToken ct = default)
CancellationToken ct = default, int maxRows = 50_000)
{
if (string.IsNullOrWhiteSpace(channel))
{
@@ -661,6 +661,9 @@ VALUES
throw new ArgumentOutOfRangeException(nameof(batchSize), batchSize, "Batch size must be > 0.");
}
if (maxRows <= 0)
throw new ArgumentOutOfRangeException(nameof(maxRows), maxRows, "Row budget must be > 0.");
var thresholdUtc = DateTime.SpecifyKind(threshold.ToUniversalTime(), DateTimeKind.Utc);
// Maintenance timeout in whole seconds; null leaves the provider default (arch-review 04 S2).
@@ -675,7 +678,7 @@ VALUES
// targeting AuditLog still trips the guard.
//
// Bounded + idempotent: DELETE TOP (@batch) caps the log/lock footprint per
// statement; the loop repeats until a batch deletes zero rows, so re-running
// statement; the loop stops at its row budget or an empty batch, so re-running
// after a crash mid-loop simply resumes. Category is the canonical
// channel-name column (e.g. 'ApiOutbound'); Action holds "{channel}.{kind}" so
// it is NOT the right column to match a bare channel name against.
@@ -698,7 +701,7 @@ VALUES
try
{
while (true)
while (totalDeleted < maxRows)
{
ct.ThrowIfCancellationRequested();
@@ -711,7 +714,7 @@ VALUES
var pBatch = cmd.CreateParameter();
pBatch.ParameterName = "@batch";
pBatch.Value = batchSize;
pBatch.Value = (int)Math.Min(batchSize, maxRows - totalDeleted);
cmd.Parameters.Add(pBatch);
var pChannel = cmd.CreateParameter();
@@ -456,6 +456,18 @@ public class TemplateEngineRepository : ITemplateEngineRepository
.ToListAsync(cancellationToken);
}
/// <inheritdoc />
public async Task<IReadOnlyList<Instance>> GetTopologyInstancesAsync(int siteId, int? areaId, int skip, int take, CancellationToken cancellationToken = default)
{
ArgumentOutOfRangeException.ThrowIfNegative(skip);
ArgumentOutOfRangeException.ThrowIfLessThan(take, 1);
ArgumentOutOfRangeException.ThrowIfGreaterThan(take, 50);
return await _context.Instances.AsNoTracking()
.Where(i => i.SiteId == siteId && i.AreaId == areaId)
.OrderBy(i => i.UniqueName).ThenBy(i => i.Id)
.Skip(skip).Take(take).ToListAsync(cancellationToken);
}
/// <inheritdoc />
public async Task<Instance?> GetInstanceByUniqueNameAsync(string uniqueName, CancellationToken cancellationToken = default)
{
@@ -675,9 +675,16 @@ public class RealOpcUaClient : IOpcUaClient
List<DataShard> touched,
CancellationToken cancellationToken)
{
foreach (var (index, nodeId, startNode) in resolved)
// Plan once for this chunk while the apply gate keeps shard occupancy stable.
var placement = MonitoredItemShardPlanner.Plan(
_dataShards.Select(s => s.ItemCount).ToArray(), ItemsPerShard, resolved.Count);
for (var itemIndex = 0; itemIndex < resolved.Count; itemIndex++)
{
var shard = await PlaceOnDataShardAsync(cancellationToken);
var (index, nodeId, startNode) = resolved[itemIndex];
var shardIndex = placement[itemIndex];
var shard = shardIndex < _dataShards.Count
? _dataShards[shardIndex]
: await CreateDataShardAsync(cancellationToken);
var handle = Guid.NewGuid().ToString();
var monitoredItem = new MonitoredItem(shard.Subscription.DefaultItem)
{
@@ -760,20 +767,6 @@ public class RealOpcUaClient : IOpcUaClient
created.Count);
}
/// <summary>
/// Returns the shard this item belongs on per
/// <see cref="MonitoredItemShardPlanner"/> — first shard with free capacity, else a
/// freshly created one. Callers must hold <see cref="_applyGate"/>.
/// </summary>
private async Task<DataShard> PlaceOnDataShardAsync(CancellationToken cancellationToken)
{
var index = MonitoredItemShardPlanner.Plan(
_dataShards.Select(s => s.ItemCount).ToList(), ItemsPerShard, 1)[0];
return index < _dataShards.Count
? _dataShards[index]
: await CreateDataShardAsync(cancellationToken);
}
/// <summary>
/// Drops a monitored item that failed to create from the shard it was placed on.
/// Callers must hold <see cref="_applyGate"/>. No ApplyChanges is issued: the server
@@ -0,0 +1,15 @@
using Microsoft.Extensions.Options;
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer;
/// <summary>Validates the MxGateway client identity before connections start.</summary>
public sealed class MxGatewayGlobalOptionsValidator : IValidateOptions<MxGatewayGlobalOptions>
{
/// <inheritdoc />
public ValidateOptionsResult Validate(string? name, MxGatewayGlobalOptions options)
{
if (string.IsNullOrWhiteSpace(options.ClientNamePrefix))
return ValidateOptionsResult.Fail("ScadaBridge:MxGateway:ClientNamePrefix must not be blank.");
return ValidateOptionsResult.Success;
}
}
@@ -0,0 +1,15 @@
using Microsoft.Extensions.Options;
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer;
/// <summary>Validates the OPC UA application identity before connections start.</summary>
public sealed class OpcUaGlobalOptionsValidator : IValidateOptions<OpcUaGlobalOptions>
{
/// <inheritdoc />
public ValidateOptionsResult Validate(string? name, OpcUaGlobalOptions options)
{
if (string.IsNullOrWhiteSpace(options.ApplicationName))
return ValidateOptionsResult.Fail("ScadaBridge:OpcUa:ApplicationName must not be blank.");
return ValidateOptionsResult.Success;
}
}
@@ -24,16 +24,18 @@ public static class ServiceCollectionExtensions
services.TryAddEnumerable(
ServiceDescriptor.Singleton<IValidateOptions<DataConnectionOptions>, DataConnectionOptionsValidator>());
// OpcUaGlobalOptions / MxGatewayGlobalOptions carry only string identity/path
// fields (no numeric or duration knobs), so there is nothing to range-validate
// — no dedicated validator is registered for them. Canonical sections are
// ScadaBridge:OpcUa / ScadaBridge:MxGateway (NF8): the bare "OpcUa"/"MxGateway"
// sections exist in no appsettings, so binding them read nothing.
// Validate identities before the first connection attempts to register with a server.
services.AddOptions<OpcUaGlobalOptions>()
.BindConfiguration("ScadaBridge:OpcUa");
.BindConfiguration("ScadaBridge:OpcUa")
.ValidateOnStart();
services.TryAddEnumerable(
ServiceDescriptor.Singleton<IValidateOptions<OpcUaGlobalOptions>, OpcUaGlobalOptionsValidator>());
services.AddOptions<MxGatewayGlobalOptions>()
.BindConfiguration("ScadaBridge:MxGateway");
.BindConfiguration("ScadaBridge:MxGateway")
.ValidateOnStart();
services.TryAddEnumerable(
ServiceDescriptor.Singleton<IValidateOptions<MxGatewayGlobalOptions>, MxGatewayGlobalOptionsValidator>());
// Register the factory for protocol extensibility
services.AddSingleton<IDataConnectionFactory, DataConnectionFactory>();
@@ -10,8 +10,19 @@ internal static class ConfigLoader
/// <summary>Deserialize the <c>appsettings.json</c> text via the source-gen context.</summary>
/// <param name="jsonText">The raw JSON text of <c>appsettings.json</c>.</param>
/// <returns>The deserialized <see cref="NotifierConfig"/>, or a default instance if deserialization yields <c>null</c>.</returns>
public static NotifierConfig Load(string jsonText) =>
JsonSerializer.Deserialize(jsonText, NotifierJsonContext.Default.NotifierConfig) ?? new NotifierConfig();
public static NotifierConfig Load(string jsonText)
{
var config = JsonSerializer.Deserialize(jsonText, NotifierJsonContext.Default.NotifierConfig) ?? new NotifierConfig();
var urls = SplitBaseUrls(config.ScadaBridge.BaseUrls);
for (var i = 0; i < urls.Length; i++)
{
if (!Uri.TryCreate(urls[i], UriKind.Absolute, out var uri)
|| (uri.Scheme != Uri.UriSchemeHttp && uri.Scheme != Uri.UriSchemeHttps)
|| string.IsNullOrWhiteSpace(uri.Host))
throw new InvalidOperationException($"ScadaBridge:BaseUrls entry {i + 1} must be an absolute HTTP(S) URI.");
}
return config;
}
/// <summary>Read and parse the <c>appsettings.json</c> sitting next to the executable.</summary>
/// <returns>The deserialized <see cref="NotifierConfig"/>.</returns>
@@ -561,18 +561,10 @@ akka {{
// 30 s timeout, so it could only add a hop and latency to every audit batch.
centralControlGrpc?.SetAuditIngestActor(auditIngest.Proxy);
// Hand the proxy to the SiteStreamGrpcServer (if registered on this node)
// so the IngestAuditEvents RPC routes incoming site batches to the singleton.
// The gRPC server is currently only registered on Site nodes; on a central
// node this resolves to null and the wiring is a no-op unless a future
// central-hosted gRPC server (or a real site→central client) is added.
var grpcServer = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer>();
grpcServer?.SetAuditIngestActor(auditIngest.Proxy);
_logger.LogInformation(
"AuditLogIngestActor singleton created (control-plane bound: {ControlBound}, "
+ "site-stream server bound: {GrpcBound})",
centralControlGrpc is not null,
grpcServer is not null);
"AuditLogIngestActor singleton created (control-plane bound: {ControlBound})",
centralControlGrpc is not null);
// Subscribe the per-site stalled
// telemetry tracker to the actor system EventStream NOW that the
@@ -121,8 +121,10 @@ public sealed class ClusterBootstrapCoordinator : IHostedService
return;
}
ct.ThrowIfCancellationRequested();
var cluster = Akka.Cluster.Cluster.Get(_system());
var addresses = order.Select(Address.Parse).ToArray();
ct.ThrowIfCancellationRequested();
cluster.JoinSeedNodes(addresses);
// Peer-first is a one-way commitment: JoinSeedNodeProcess never self-forms. If the founder
@@ -184,7 +186,7 @@ public sealed class ClusterBootstrapCoordinator : IHostedService
if (await TryConnectAsync(host, port, ct).ConfigureAwait(false)) return true;
if (sw.Elapsed >= window) return false;
try { await Task.Delay(interval, ct).ConfigureAwait(false); }
catch (OperationCanceledException) { return false; }
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
}
return false;
}
@@ -29,12 +29,13 @@ namespace ZB.MOM.WW.ScadaBridge.Host.Health;
/// whole check stays cheap and readiness polling stays fast.
/// </para>
/// <para>
/// <b>Required-always vs if-applicable.</b> All five central singleton proxies are
/// created unconditionally on a central node (there is no feature/config gate around
/// any of them), so all five are treated as required-always here. If a future
/// singleton is created behind a feature flag, it should NOT be added to
/// <see cref="RequiredSingletonProxyNames"/> — "if applicable" means skip when its
/// feature is off.
/// <b>Required versus best-effort.</b> Seven central singleton proxies are created
/// unconditionally. The five listed in <see cref="RequiredSingletonProxyNames"/>
/// are required for readiness. The KPI-history recorder and pending-deployment purge
/// are deliberately excluded: they provide best-effort observability and hygiene,
/// respectively, and their availability is not a correctness prerequisite. Future
/// additions must be classified by readiness impact, not just unconditional creation;
/// a feature-gated singleton must also be skipped when its feature is disabled.
/// </para>
/// <para>
/// <b>Failover flakiness.</b> During a brief singleton handover the singleton may be
@@ -60,7 +61,7 @@ public sealed class RequiredSingletonsHealthCheck : IHealthCheck
/// <summary>
/// Local actor names (under <c>/user</c>) of the <c>ClusterSingletonProxy</c>
/// actors for the singletons that must always be running on a central node.
/// Matches the unconditional proxy registrations in
/// Selects the five required services from the seven unconditional registrations in
/// <c>AkkaHostedService.RegisterCentralActors</c>.
/// </summary>
public static readonly IReadOnlyList<string> RequiredSingletonProxyNames = new[]
@@ -230,6 +230,8 @@ try
builder.WebHost.ConfigureKestrel(options =>
{
options.Limits.Http2.KeepAlivePingDelay = SiteKestrelSetup.Http2KeepAlivePingDelay;
options.Limits.Http2.KeepAlivePingTimeout = SiteKestrelSetup.Http2KeepAlivePingTimeout;
// The HTTP/1.1 (+ HTTP/2) surface from the configured URLs, re-declared so it
// survives the explicit-endpoints switch above.
foreach (var httpPort in httpPorts)
@@ -28,7 +28,7 @@ namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
/// active (a transient during failover) bounces at most one hop, never ping-pongs.</para>
///
/// <para><b>Fidelity.</b> Method, path, query string, body and headers are relayed
/// verbatim (hop-by-hop headers and <c>Host</c> excepted), so the active node
/// verbatim (hop-by-hop headers, <c>Host</c> and <c>Accept-Encoding</c> excepted), so the active node
/// authenticates the caller's own <c>Authorization</c>/<c>X-API-Key</c> credential and
/// applies its own body cap and method timeout. The active node's status, content type
/// and body are replayed to the caller unchanged. The body is buffered once on the
@@ -61,6 +61,8 @@ public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
{
"Host", "Content-Length", "Transfer-Encoding", "Connection", "Keep-Alive",
"Proxy-Connection", "Upgrade", "TE", "Trailer", "Expect",
// The relay copies raw response bytes and Content-Type, not Content-Encoding.
"Accept-Encoding",
};
private readonly IHttpClientFactory _httpClientFactory;
@@ -16,6 +16,9 @@ public sealed class KpiHistoryOptions
/// </summary>
public TimeSpan SampleInterval { get; set; } = TimeSpan.FromSeconds(60);
/// <summary>Maximum wait for one KPI source (default 10 seconds). Unfinished sources remain owned and are skipped on later ticks.</summary>
public TimeSpan SourceTimeout { get; set; } = TimeSpan.FromSeconds(10);
/// <summary>
/// Central retention window in days for recorded KPI samples (default 90,
/// range <c>[1, 3650]</c>). Rows older than this are dropped by the purge
@@ -46,6 +46,9 @@ public sealed class KpiHistoryOptionsValidator : OptionsValidatorBase<KpiHistory
/// <inheritdoc />
protected override void Validate(ValidationBuilder builder, KpiHistoryOptions options)
{
builder.RequireThat(options.SourceTimeout > TimeSpan.Zero && options.SourceTimeout <= TimeSpan.FromDays(1),
$"ScadaBridge:KpiHistory:{nameof(KpiHistoryOptions.SourceTimeout)} must be > 0 and <= 1 day.");
builder.RequireThat(options.SampleInterval > TimeSpan.Zero,
$"ScadaBridge:KpiHistory:{nameof(KpiHistoryOptions.SampleInterval)} ({options.SampleInterval}) " +
"must be > 0; it is the recorder's sampling cadence.");
@@ -58,9 +58,10 @@ namespace ZB.MOM.WW.ScadaBridge.KpiHistory;
/// <para>
/// <b>DI scopes.</b> <see cref="IKpiSampleSource"/>s and
/// <see cref="IKpiHistoryRepository"/> are scoped EF Core-backed services; the
/// recorder opens a fresh <see cref="IServiceScope"/> per tick and resolves
/// there, asynchronously disposing it (<c>CreateAsyncScope</c>) so EF Core's
/// async connection cleanup does not block.
/// recorder opens a fresh <see cref="IServiceScope"/> for the tick's repository
/// and a separate scope for each source collection. A timed-out collection owns
/// its scope until actual completion, while later sources use separate DbContexts.
/// Scopes are asynchronously disposed (<c>CreateAsyncScope</c>).
/// </para>
/// <para>
/// <b>Singleton wiring.</b> The <c>Props</c> is built in the Host on the
@@ -87,6 +88,8 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
/// </summary>
private static readonly TimeSpan BackfillInitialDelay = TimeSpan.FromSeconds(15);
// Shared by every recorder incarnation on this root provider.
private readonly KpiSourceAdmission _sourceAdmission;
private readonly IServiceProvider _serviceProvider;
private readonly KpiHistoryOptions _options;
private readonly ILogger<KpiHistoryRecorderActor> _logger;
@@ -176,6 +179,7 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
ILogger<KpiHistoryRecorderActor> logger)
{
_serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
_sourceAdmission = serviceProvider.GetRequiredService<KpiSourceAdmission>();
_options = options ?? throw new ArgumentNullException(nameof(options));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -307,9 +311,8 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
}
/// <summary>
/// Runs a single sampling pass: opens a DI scope, enumerates every registered
/// <see cref="IKpiSampleSource"/>, collects each source's samples (isolating per-source
/// faults), and bulk-writes the combined batch. The whole body is wrapped so the
/// Runs a single sampling pass: enumerates every registered <see cref="IKpiSampleSource"/>,
/// waits a bounded time for each separately scoped collection, and bulk-writes the combined batch. The whole body is wrapped so the
/// returned task never faults — best-effort observability must never disrupt anything.
/// </summary>
private async Task RunSamplePass(DateTime capturedAt, CancellationToken cancellationToken)
@@ -317,17 +320,32 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
try
{
await using var scope = _serviceProvider.CreateAsyncScope();
var sources = scope.ServiceProvider.GetServices<IKpiSampleSource>();
var sources = scope.ServiceProvider.GetServices<IKpiSampleSource>().ToArray();
var repository = scope.ServiceProvider.GetRequiredService<IKpiHistoryRepository>();
var samples = new List<KpiSample>();
foreach (var source in sources)
for (var sourceIndex = 0; sourceIndex < sources.Length; sourceIndex++)
{
var sourceName = sources[sourceIndex].Source;
// Each call owns a separate scope, so a cancellation-ignoring source
// neither shares a busy DbContext with healthy sources nor loses its scope.
var index = sourceIndex;
if (!_sourceAdmission.TryStart(index,
() => CollectSourceAsync(index, capturedAt, cancellationToken), out var collection))
continue;
_ = collection.ContinueWith(t =>
_logger.LogError(t.Exception, "KPI source {Source} collection failed.", sourceName),
CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
try
{
var collected = await source.CollectAsync(capturedAt, cancellationToken);
var collected = await collection.WaitAsync(_options.SourceTimeout, cancellationToken);
samples.AddRange(collected);
}
catch (TimeoutException)
{
_logger.LogWarning("KPI source {Source} exceeded its collection deadline; unfinished work remains owned and will not be restarted.", sourceName);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// Shutdown interrupted collection; abandon the rest of the pass. The next
@@ -337,7 +355,7 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
catch (Exception ex)
{
// A throwing source must NOT abort the pass or the other sources.
_logger.LogError(ex, "KPI source {Source} failed to collect samples.", source.Source);
_logger.LogError(ex, "KPI source {Source} failed to collect samples.", sourceName);
}
}
@@ -367,6 +385,17 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
}
}
/// <summary>Owns a source's scope and cancellation source until its actual collection completes, even after the caller stops waiting.</summary>
private async Task<IReadOnlyList<KpiSample>> CollectSourceAsync(
int sourceIndex, DateTime capturedAt, CancellationToken cancellationToken)
{
await using var scope = _serviceProvider.CreateAsyncScope();
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
deadline.CancelAfter(_options.SourceTimeout);
var source = scope.ServiceProvider.GetServices<IKpiSampleSource>().ElementAt(sourceIndex);
return await source.CollectAsync(capturedAt, deadline.Token).ConfigureAwait(false);
}
/// <summary>
/// Handles a purge tick: computes both retention cut-offs on the actor thread (raw samples
/// at <see cref="KpiHistoryOptions.RetentionDays"/>, hourly rollups at the longer
@@ -0,0 +1,33 @@
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi;
namespace ZB.MOM.WW.ScadaBridge.KpiHistory;
/// <summary>
/// Root-provider-owned admission for source collections. Actor replacement must not
/// start another call while an earlier cancellation-ignoring collection still owns its scope.
/// Registration indices are stable because the root provider's service collection is immutable.
/// </summary>
internal sealed class KpiSourceAdmission
{
private readonly object _gate = new();
private readonly Dictionary<int, Task<IReadOnlyList<KpiSample>>> _tasks = new();
internal bool TryStart(
int registrationIndex,
Func<Task<IReadOnlyList<KpiSample>>> collect,
out Task<IReadOnlyList<KpiSample>> collection)
{
lock (_gate)
{
if (_tasks.TryGetValue(registrationIndex, out var existing) && !existing.IsCompleted)
{
collection = existing;
return false;
}
collection = Task.Run(collect);
_tasks[registrationIndex] = collection;
return true;
}
}
}
@@ -1,5 +1,6 @@
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using ZB.MOM.WW.Configuration;
namespace ZB.MOM.WW.ScadaBridge.KpiHistory;
@@ -22,7 +23,8 @@ public static class ServiceCollectionExtensions
/// <summary>
/// Registers the KPI History component services: the validated
/// <see cref="KpiHistoryOptions"/> binding. Idempotent re-registration of the
/// <see cref="KpiHistoryOptions"/> binding and provider-owned source admission.
/// Idempotent re-registration of the
/// validator is handled by the shared <c>AddValidatedOptions</c> helper
/// (<c>TryAddEnumerable</c>); the options binding itself is call-once per
/// <see cref="IServiceCollection"/>.
@@ -38,6 +40,7 @@ public static class ServiceCollectionExtensions
// Binds the "ScadaBridge:KpiHistory" section, registers the validator,
// and enables ValidateOnStart in one call — same shape as AddAuditLog.
services.AddValidatedOptions<KpiHistoryOptions, KpiHistoryOptionsValidator>(configuration, OptionsSection);
services.TryAddSingleton<KpiSourceAdmission>();
return services;
}
@@ -60,7 +60,10 @@ public sealed class FlowDebugRunner
/// loop running on the node after the caller's Ask has already expired, holding its scoped
/// provider and pulling from the site for a result nobody can receive.
/// </summary>
public const int MaxWaitSeconds = 300;
public const int MaxWaitSeconds = 295;
/// <summary>Time reserved for serializing and returning a management reply.</summary>
public static readonly TimeSpan ResponseGrace = TimeSpan.FromSeconds(5);
/// <summary>
/// Arch-review F2: <see cref="InboundScriptExecutor"/>'s code for a routed call refused
@@ -83,22 +86,8 @@ public sealed class FlowDebugRunner
public static readonly TimeSpan SettleWindow = TimeSpan.FromSeconds(2);
/// <summary>
/// Row cap asked of the site on each <see cref="IPullTraceEventsClient.PullAsync"/> poll — the
/// site's own ceiling (<c>SiteStreamGrpcServer.MaxTraceEventsPerPull</c> = 2000; a larger
/// <c>max_events</c> is rejected <c>InvalidArgument</c>), so this asks for everything the site
/// will serve.
///
/// <para>
/// Final review I1: this was 500, which was ample before step logs but is not any more. The
/// site query is <c>ORDER BY OccurredAtUtc ASC LIMIT</c>, so the page is the OLDEST rows in
/// the window — and a single traced run now emits up to 201 <c>ScriptLog</c> rows (design
/// §3, <c>MaxLogLinesPerRun</c> + the closing dropped-count row). Two chatty runs could
/// therefore fill a 500-row page and push a later spawned run's <c>ScriptRun</c> row past the
/// end of it, making <see cref="WriteAndPollAsync"/> report a false
/// <see cref="NoTriggerCode"/> for a run that did happen. 2000 makes that ~10x harder; the
/// real fix (paging by last-seen timestamp, or excluding <c>ScriptLog</c> from the awaited-run
/// poll) is deferred-work register row 117.
/// </para>
/// Site page ceiling. Full pages are drained with an exclusive timestamp/event-id cursor;
/// execution IDs are split into independent batches of at most 64.
/// </summary>
public const int PullTraceEventsMax = 2000;
@@ -136,8 +125,12 @@ public sealed class FlowDebugRunner
/// registered none (a test composition root) collects raw but is still capped.
/// </para>
/// </summary>
public async Task<FlowDebugRunResult> RunMethodAsync(
public Task<FlowDebugRunResult> RunMethodAsync(
IServiceProvider sp, FlowDebugRunMethodCommand cmd, AuthenticatedUser user, CancellationToken ct)
=> WithinCommandBudgetAsync(sp, (deadline, token) => RunMethodCoreAsync(sp, cmd, user, deadline, token), ct);
private async Task<FlowDebugRunResult> RunMethodCoreAsync(
IServiceProvider sp, FlowDebugRunMethodCommand cmd, AuthenticatedUser user, DateTimeOffset commandDeadline, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(sp);
ArgumentNullException.ThrowIfNull(cmd);
@@ -240,8 +233,12 @@ public sealed class FlowDebugRunner
/// Live run of a value-change trigger (design §5.3/§5.4): traced routed write through the
/// site's normal path, then poll the site's trace rows for the spawned run(s).
/// </summary>
public async Task<FlowDebugRunResult> RunValueChangeAsync(
public Task<FlowDebugRunResult> RunValueChangeAsync(
IServiceProvider sp, FlowDebugRunValueChangeCommand cmd, AuthenticatedUser user, CancellationToken ct)
=> WithinCommandBudgetAsync(sp, (deadline, token) => RunValueChangeCoreAsync(sp, cmd, user, deadline, token), ct);
private async Task<FlowDebugRunResult> RunValueChangeCoreAsync(
IServiceProvider sp, FlowDebugRunValueChangeCommand cmd, AuthenticatedUser user, DateTimeOffset commandDeadline, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(sp);
ArgumentNullException.ThrowIfNull(cmd);
@@ -259,7 +256,7 @@ public sealed class FlowDebugRunner
?? throw new ManagementCommandException($"Site with ID {instance.SiteId} not found.", NotFoundCode);
return await WriteAndPollAsync(
sp, user, instance, site, cmd.AttributeName, cmd.Value, cmd.WaitSeconds, triggerScript: null, ct)
sp, user, instance, site, cmd.AttributeName, cmd.Value, cmd.WaitSeconds, triggerScript: null, commandDeadline, ct)
.ConfigureAwait(false);
}
@@ -272,8 +269,12 @@ public sealed class FlowDebugRunner
/// rather than through <see cref="RouteHelper"/> is deliberate: the helper emits its own
/// always-on RouteCall row with no operator actor, which would give the graph two roots.
/// </summary>
public async Task<FlowDebugRunResult> RunScriptAsync(
public Task<FlowDebugRunResult> RunScriptAsync(
IServiceProvider sp, FlowDebugRunScriptCommand cmd, AuthenticatedUser user, CancellationToken ct)
=> WithinCommandBudgetAsync(sp, (deadline, token) => RunScriptCoreAsync(sp, cmd, user, deadline, token), ct);
private async Task<FlowDebugRunResult> RunScriptCoreAsync(
IServiceProvider sp, FlowDebugRunScriptCommand cmd, AuthenticatedUser user, DateTimeOffset commandDeadline, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(sp);
ArgumentNullException.ThrowIfNull(cmd);
@@ -312,7 +313,7 @@ public sealed class FlowDebugRunner
if (cmd.Value is null)
throw new ManagementCommandException($"Value is required to fire '{script.CanonicalName}' (writes '{attribute}').", ValidationFailedCode);
return await WriteAndPollAsync(
sp, user, instance, site, attribute!, cmd.Value, cmd.WaitSeconds, script.CanonicalName, ct).ConfigureAwait(false);
sp, user, instance, site, attribute!, cmd.Value, cmd.WaitSeconds, script.CanonicalName, commandDeadline, ct).ConfigureAwait(false);
}
var resolveRef = await SchemaRefResolver.BuildAsync(
@@ -431,7 +432,7 @@ public sealed class FlowDebugRunner
/// </summary>
private async Task<FlowDebugRunResult> WriteAndPollAsync(
IServiceProvider sp, AuthenticatedUser user, Instance instance, Site site,
string attributeName, string? value, int waitSeconds, string? triggerScript, CancellationToken ct)
string attributeName, string? value, int waitSeconds, string? triggerScript, DateTimeOffset commandDeadline, CancellationToken ct)
{
var root = Guid.NewGuid();
var startedAt = DateTimeOffset.UtcNow;
@@ -496,22 +497,53 @@ public sealed class FlowDebugRunner
var puller = sp.GetRequiredService<IPullTraceEventsClient>();
var wait = Math.Clamp(waitSeconds, 1, MaxWaitSeconds);
var deadline = _now() + TimeSpan.FromSeconds(wait);
if (deadline > commandDeadline) deadline = commandDeadline;
var known = new Dictionary<Guid, AuditEvent>();
DateTimeOffset? lastNew = null;
var since = startedAt.UtcDateTime.AddSeconds(-5);
// Root is a fresh GUID for this command, so matching it and its discovered descendants
// requires no cross-node timestamp lower bound. Site clocks may be behind central.
var since = DateTime.MinValue;
// Stable ID batches avoid silently dropping descendants beyond the per-read ID cap.
var assigned = new HashSet<Guid>();
var batches = new List<TraceReadBatch>();
while (!ct.IsCancellationRequested)
{
var ids = new HashSet<Guid> { root };
var discovered = new HashSet<Guid> { root };
foreach (var e in known.Values)
{
var d = AuditDetailsCodec.Deserialize(e.DetailsJson);
if (d.ExecutionId is { } x) ids.Add(x);
if (d.ExecutionId is { } x) discovered.Add(x);
}
var pulled = await puller.PullAsync(site.SiteIdentifier, ids.Take(64).ToList(), since, PullTraceEventsMax, ct).ConfigureAwait(false);
foreach (var ids in discovered.Where(id => assigned.Add(id)).Chunk(64))
batches.Add(new TraceReadBatch(ids, since));
var added = false;
foreach (var e in pulled)
foreach (var batch in batches)
{
if (known.TryAdd(e.EventId, e)) added = true;
// ScriptRun/operation closing rows are inserted at completion but stamped with
// their start time, potentially behind a cursor that already saw live log rows.
// Drain every page within this scan, then intentionally rescan on the next poll.
// This trades repeated transfer for correct late/backdated-row discovery.
batch.SinceUtc = since;
batch.AfterEventId = null;
while (!ct.IsCancellationRequested && _now() < deadline)
{
var pulled = await puller.PullAsync(site.SiteIdentifier, batch.ExecutionIds,
batch.SinceUtc, PullTraceEventsMax, ct, batch.AfterEventId).ConfigureAwait(false);
foreach (var e in pulled)
if (known.TryAdd(e.EventId, e)) added = true;
if (pulled.Count == 0) break;
var last = pulled[^1];
var nextTime = last.OccurredAtUtc.UtcDateTime;
var nextId = last.EventId.ToString("D");
// An old peer ignoring the additive cursor must not cause a tight repeat loop.
var advanced = nextTime > batch.SinceUtc || nextTime == batch.SinceUtc
&& (batch.AfterEventId is null || string.CompareOrdinal(nextId, batch.AfterEventId) > 0);
if (!advanced) break;
batch.SinceUtc = nextTime;
batch.AfterEventId = nextId;
if (pulled.Count < PullTraceEventsMax) break;
}
}
if (added) lastNew = _now();
@@ -522,8 +554,11 @@ public sealed class FlowDebugRunner
await _delay(PollInterval, ct).ConfigureAwait(false);
}
var ordered = new List<AuditEvent> { redactedRoot };
ordered.AddRange(known.Values.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId));
var collector = new FlowTraceCollector();
collector.Add(redactedRoot);
foreach (var evt in known.Values.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId))
collector.Add(evt, lowPriority: FlowStepProjector.IsLogLine(evt));
var ordered = collector.Complete();
var completed = known.Values.Any(IsAwaitedRun);
return new FlowDebugRunResult(
root, completed,
@@ -531,7 +566,7 @@ public sealed class FlowDebugRunner
? $"No traced run was spawned within {wait}s (no value-change trigger on '{attributeName}', or no device echo)."
: $"No traced run of '{triggerScript}' was spawned within {wait}s after writing '{attributeName}' (trigger condition not met, or no device echo).",
completed ? null : NoTriggerCode,
ordered, false, startedAt, DateTimeOffset.UtcNow);
ordered, collector.Truncated, startedAt, DateTimeOffset.UtcNow, LogsTruncated: collector.LogsTruncated);
}
/// <summary>Replay (design §7.1): the stored ScriptTrace chain for any execution id.</summary>
@@ -590,6 +625,34 @@ public sealed class FlowDebugRunner
}
}
private sealed class TraceReadBatch(Guid[] executionIds, DateTime sinceUtc)
{
public Guid[] ExecutionIds { get; } = executionIds;
public DateTime SinceUtc { get; set; } = sinceUtc;
public string? AfterEventId { get; set; }
}
private async Task<FlowDebugRunResult> WithinCommandBudgetAsync(
IServiceProvider sp, Func<DateTimeOffset, CancellationToken, Task<FlowDebugRunResult>> run, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(sp);
var timeout = sp.GetService<Microsoft.Extensions.Options.IOptions<ManagementServiceOptions>>()?.Value.LongRunningCommandTimeout
?? TimeSpan.FromMinutes(5);
var budget = timeout - ResponseGrace;
if (budget <= TimeSpan.Zero)
throw new ManagementCommandException("Long-running command timeout must exceed the response grace.", ValidationFailedCode);
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
deadline.CancelAfter(budget);
try
{
return await run(_now() + budget, deadline.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested)
{
throw new ManagementCommandException("Flow Debugger command exceeded its total execution budget.", TimeoutCode);
}
}
private static void EnsureActiveNode(IServiceProvider sp)
{
var gate = sp.GetService<IActiveNodeGate>();
@@ -651,6 +714,10 @@ public sealed class FlowDebugRunner
{
await writer.WriteAsync(evt, ct).ConfigureAwait(false);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
throw;
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Flow Debugger central audit write failed for {Action}", evt.Action);
@@ -730,7 +730,7 @@ public class ManagementActor : ReceiveActor
private static List<T> Page<T>(IEnumerable<T> source, int skip, int? take) =>
source
.Skip(Math.Max(0, skip))
.Take(take is > 0 and <= 1000 ? take.Value : int.MaxValue)
.Take(take is > 0 ? Math.Min(take.Value, 1000) : int.MaxValue)
.ToList();
private static async Task<object?> HandleGetTemplate(IServiceProvider sp, GetTemplateCommand cmd)
@@ -23,8 +23,8 @@ public sealed class ManagementServiceOptionsValidator : OptionsValidatorBase<Man
$"ScadaBridge:ManagementService:CommandTimeout must be a positive duration " +
$"(was {options.CommandTimeout}); it is the Ask timeout for management commands.");
builder.RequireThat(options.LongRunningCommandTimeout > TimeSpan.Zero,
$"ScadaBridge:ManagementService:LongRunningCommandTimeout must be a positive duration " +
builder.RequireThat(options.LongRunningCommandTimeout > FlowDebug.FlowDebugRunner.ResponseGrace,
$"ScadaBridge:ManagementService:LongRunningCommandTimeout must exceed the 5-second response grace " +
$"(was {options.LongRunningCommandTimeout}); it is the Ask timeout for transport/deploy commands.");
}
}
@@ -33,8 +33,8 @@ public sealed class SiteCallAuditOptionsValidator : OptionsValidatorBase<SiteCal
$"ScadaBridge:SiteCallAudit:ReconciliationInterval must be a positive duration " +
$"(was {options.ReconciliationInterval}); it is the period of the per-site reconciliation pull.");
builder.RequireThat(options.ReconciliationBatchSize > 0,
$"ScadaBridge:SiteCallAudit:ReconciliationBatchSize must be positive " +
builder.RequireThat(options.ReconciliationBatchSize is > 0 and <= Communication.Grpc.SiteStreamGrpcServer.MaxReconciliationRowsPerPull,
$"ScadaBridge:SiteCallAudit:ReconciliationBatchSize must be 1..2000 " +
$"(was {options.ReconciliationBatchSize}); it is the max rows requested per PullSiteCalls RPC.");
builder.RequireThat(options.PurgeInterval > TimeSpan.Zero,
@@ -279,6 +279,9 @@ public class AlarmActor : ReceiveActor
/// </summary>
private void HandleAttributeValueChanged(AttributeValueChanged changed)
{
// Alarm audit rows have no inferred-edge metadata: never assert that an armed
// device publish was caused by the candidate writer. Observed static writes retain it.
var observedSource = changed.SourceExecutionInferred ? null : changed.SourceExecutionId;
// Expression triggers evaluate against a snapshot of every attribute,
// not a single monitored attribute. Keep the snapshot current for every
// change before the IsMonitoredAttribute gate (which does not apply).
@@ -288,7 +291,7 @@ public class AlarmActor : ReceiveActor
// (ParentExecutionId tag-cascade): remember which
// execution (if any) wrote the newest value folded into the snapshot,
// so the off-dispatcher evaluation can attribute its raise to it.
_latestSourceExecutionId = changed.SourceExecutionId;
_latestSourceExecutionId = observedSource;
}
else if (!IsMonitoredAttribute(changed.AttributeName))
{
@@ -300,7 +303,7 @@ public class AlarmActor : ReceiveActor
{
if (_triggerType == AlarmTriggerType.HiLo)
{
HandleHiLoTransition(EvaluateHiLo(changed.Value), changed.SourceExecutionId);
HandleHiLoTransition(EvaluateHiLo(changed.Value), observedSource);
return;
}
@@ -322,7 +325,7 @@ public class AlarmActor : ReceiveActor
_ => false
};
ApplyTriggeredState(isTriggered, changed.SourceExecutionId);
ApplyTriggeredState(isTriggered, observedSource);
}
catch (Exception ex)
{
@@ -174,7 +174,7 @@ public class InstanceActor : ReceiveActor
/// <c>Trace</c>, and the arm is consumed. Bounded by the attribute count (one entry per
/// attribute, newest wins); no sweep — an unconsumed arm is overwritten or expires in place.
/// Armed at ISSUE, disarmed on a failed/timed-out write (arch-review F6, <see cref="DisarmTrace"/>);
/// a successful write's echo consumes it.
/// the first subsequent device update consumes it, with explicitly inferred linkage.
/// </summary>
private readonly Dictionary<string, (Guid SourceExecutionId, DateTimeOffset ExpiresAt)> _traceArms = new();
// Maps each tag path to every attribute canonical name that references it.
@@ -337,7 +337,7 @@ public class InstanceActor : ReceiveActor
Receive<RetryTagSubscribe>(msg => SendTagSubscribe(msg.ConnectionName));
// Arch-review F6: a traced write the DCL rejected (or that timed out) drops its own arm,
// so an unrelated later update is not stamped as that write's device echo.
// so a later update is not associated with that failed write.
Receive<DisarmTrace>(msg =>
{
if (_traceArms.TryGetValue(msg.AttributeName, out var arm)
@@ -751,10 +751,10 @@ public class InstanceActor : ReceiveActor
DateTimeOffset.UtcNow);
// Flow Debugger trace arm (design §5.3): only a TRACED write with a known writer arms;
// the device echo that follows is stamped in HandleTagValueUpdate. Untraced writes
// the next device update is marked as inferred in HandleTagValueUpdate. Untraced writes
// never touch the map, so production behaviour is unchanged. The arm is taken at ISSUE
// and dropped again on a failed/timed-out write (arch-review F6, via the DisarmTrace
// self-message below); a successful write's echo consumes it.
// self-message below); the first subsequent device update consumes it, with explicitly inferred linkage.
Guid? armed = null;
if (command.Trace && command.SourceExecutionId is { } traceSource)
{
@@ -1063,9 +1063,9 @@ public class InstanceActor : ReceiveActor
// O(1) lookup off the hot ingest path (was a linear FirstOrDefault).
_resolvedAttributeByName.TryGetValue(attrName, out var resolved);
// Flow Debugger trace arm (design §5.3): consume the arm for this attribute — the
// echo is stamped whatever its value (a device that clamps or rejects the write
// still shows what it did) — unless the arm has already expired.
// A device subscription has no write-request identity. Associate only the first
// update inside the arm window, explicitly as inferred: a periodic publish may
// win this race, and equality cannot establish causality (devices can coerce values).
Guid? traceSource = null;
var traced = false;
if (_traceArms.Remove(attrName, out var arm) && arm.ExpiresAt > DateTimeOffset.UtcNow)
@@ -1085,7 +1085,7 @@ public class InstanceActor : ReceiveActor
HandleAttributeValueChanged(new AttributeValueChanged(
_instanceUniqueName, update.TagPath, attrName,
typedList, QualityToString(update.Quality), update.Timestamp,
SourceExecutionId: traceSource, Trace: traced));
SourceExecutionId: traceSource, Trace: traced, SourceExecutionInferred: traced));
}
else
{
@@ -1103,7 +1103,7 @@ public class InstanceActor : ReceiveActor
PublishAndNotifyChildren(new AttributeValueChanged(
_instanceUniqueName, update.TagPath, attrName,
currentValue, "Bad", update.Timestamp,
SourceExecutionId: traceSource, Trace: traced), evaluateWaiters: false);
SourceExecutionId: traceSource, Trace: traced, SourceExecutionInferred: traced), evaluateWaiters: false);
}
continue;
}
@@ -1118,7 +1118,7 @@ public class InstanceActor : ReceiveActor
HandleAttributeValueChanged(new AttributeValueChanged(
_instanceUniqueName, update.TagPath, attrName,
value, QualityToString(update.Quality), update.Timestamp,
SourceExecutionId: traceSource, Trace: traced));
SourceExecutionId: traceSource, Trace: traced, SourceExecutionInferred: traced));
}
}
@@ -303,7 +303,7 @@ public class ScriptActor : ReceiveActor, IWithTimers
TriggerKind = "valuechange",
TriggerAttribute = changed.AttributeName,
TriggerValue = AttributeValueCodec.Encode(changed.Value),
TriggerSource = "value-change",
TriggerSource = changed.SourceExecutionInferred ? "inferred-device-update" : "value-change",
});
}
else
@@ -64,7 +64,7 @@ public sealed class ScriptExecutionScheduler : TaskScheduler, IDisposable
/// </summary>
internal const int InstancesPerScriptThread = 8;
/// <summary>Thread-name prefix; also the inlining guard in <see cref="TryExecuteTaskInline"/>.</summary>
/// <summary>Thread-name prefix for diagnostics; inlining is guarded by scheduler identity.</summary>
private const string ThreadNamePrefix = "script-execution-";
private readonly BlockingCollection<Task> _queue = new();
@@ -338,8 +338,12 @@ public sealed class ScriptExecutionScheduler : TaskScheduler, IDisposable
}
}
[ThreadStatic]
private static ScriptExecutionScheduler? CurrentScheduler;
private void WorkerLoop(int index)
{
CurrentScheduler = this;
CurrentWorkerSlot = index;
var slot = _slots[index];
try
@@ -375,6 +379,7 @@ public sealed class ScriptExecutionScheduler : TaskScheduler, IDisposable
finally
{
CurrentWorkerSlot = null;
CurrentScheduler = null;
}
}
@@ -386,7 +391,7 @@ public sealed class ScriptExecutionScheduler : TaskScheduler, IDisposable
{
// Only inline if we are already on one of this scheduler's worker threads,
// so script work never escapes onto a thread-pool thread.
if (Thread.CurrentThread.Name?.StartsWith(ThreadNamePrefix, StringComparison.Ordinal) != true)
if (!ReferenceEquals(CurrentScheduler, this))
return false;
return TryExecuteTask(task);
@@ -411,7 +411,7 @@ public class ScriptRuntimeContext
{
// The collector is a debug-only observer: it must never fail a script call, and it
// must never keep the durable audit row from reaching the writer (review, 2026-09-06).
_logger.LogWarning(ex, "Flow trace collector rejected {Action} ({EventId})", evt.Action, evt.EventId);
_innerLogger.LogWarning(ex, "Flow trace collector rejected {Action} ({EventId})", evt.Action, evt.EventId);
}
}
@@ -548,7 +548,7 @@ public class ScriptRuntimeContext
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Flow trace event build failed for {Kind}", kind);
_innerLogger.LogWarning(ex, "Flow trace event build failed for {Kind}", kind);
return;
}
@@ -560,7 +560,7 @@ public class ScriptRuntimeContext
if (!t.IsCompletedSuccessfully)
{
_ = t.ContinueWith(
x => _logger.LogWarning(x.Exception, "Flow trace write failed for {Kind}", kind),
x => _innerLogger.LogWarning(x.Exception, "Flow trace write failed for {Kind}", kind),
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
@@ -568,7 +568,7 @@ public class ScriptRuntimeContext
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Flow trace write failed for {Kind}", kind);
_innerLogger.LogWarning(ex, "Flow trace write failed for {Kind}", kind);
}
}
@@ -1057,17 +1057,22 @@ public class ScriptRuntimeContext
return result.ReturnValue;
}
/// <summary>
/// Provides access to shared script execution via the Scripts property.
/// </summary>
public ScriptCallHelper Scripts => new(_sharedScriptLibrary, this, _currentCallDepth, _maxCallDepth, _logger);
private ScriptCallHelper? _scripts;
private ExternalSystemHelper? _externalSystem;
private DatabaseHelper? _database;
private NotifyHelper? _notify;
private AlarmsAccessor? _alarms;
private TrackingHelper? _tracking;
/// <summary>Provides access to shared script execution via a helper cached for this context.</summary>
public ScriptCallHelper Scripts => _scripts ??= new(_sharedScriptLibrary, this, _currentCallDepth, _maxCallDepth, _logger);
/// <summary>
/// Provides access to external system calls.
/// ExternalSystem.Call("systemName", "methodName", params)
/// ExternalSystem.CachedCall("systemName", "methodName", params)
/// </summary>
public ExternalSystemHelper ExternalSystem => new(
public ExternalSystemHelper ExternalSystem => _externalSystem ??= new(
_externalSystemClient, _instanceName, _logger, _executionId, _auditWriter, _siteId, _sourceScript,
// Emit CachedSubmit telemetry
// on every ExternalSystem.CachedCall enqueue.
@@ -1089,7 +1094,7 @@ public class ScriptRuntimeContext
/// Database.Connection("name")
/// Database.CachedWrite("name", "sql", params)
/// </summary>
public DatabaseHelper Database => new(
public DatabaseHelper Database => _database ??= new(
_databaseGateway,
_instanceName,
_logger,
@@ -1127,7 +1132,7 @@ public class ScriptRuntimeContext
/// Best-effort per alog.md §7 — a thrown writer never aborts the script's
/// <c>Send</c>.
/// </remarks>
public NotifyHelper Notify => new(
public NotifyHelper Notify => _notify ??= new(
_storeAndForward, _siteCommunicationActor, _siteId, _instanceName, _sourceScript, _askTimeout, _logger,
_executionId, _auditWriter,
// (ParentExecutionId): the spawning execution's id,
@@ -1146,7 +1151,7 @@ public class ScriptRuntimeContext
/// Gateway) conditions alike — so a <c>Call</c> script can answer "what is in alarm
/// right now?" without an on-trigger <c>Alarm</c> context.
/// </summary>
public AlarmsAccessor Alarms => new(this);
public AlarmsAccessor Alarms => _alarms ??= new(this);
/// <summary>
/// Backing read for <see cref="AlarmsAccessor.CurrentAsync"/>: Asks THIS instance's
@@ -1210,7 +1215,7 @@ public class ScriptRuntimeContext
/// returns a <see cref="TrackingStatusSnapshot"/>, or <c>null</c> when the
/// id is unknown / has already been purged.
/// </summary>
public TrackingHelper Tracking => new(_operationTrackingStore, _logger);
public TrackingHelper Tracking => _tracking ??= new(_operationTrackingStore, _logger);
/// <summary>
/// Helper class for Scripts.CallShared() syntax.
@@ -28,6 +28,7 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
public sealed class TriggerEvalGate : IDisposable
{
private readonly SemaphoreSlim _gate;
private bool _processOwned;
private static volatile TriggerEvalGate? _shared;
private static readonly object SharedLock = new();
@@ -48,7 +49,8 @@ public sealed class TriggerEvalGate : IDisposable
/// <summary>
/// The process-wide gate, used when no gate is injected. Lazily created from
/// <see cref="SiteRuntimeOptions.TriggerEvalMaxConcurrency"/>; the first caller wins.
/// <see cref="SiteRuntimeOptions.TriggerEvalMaxConcurrency"/>; the first caller wins. The shared gate is process-owned; calling
/// <see cref="Dispose"/> on it does not release the shared semaphore.
/// </summary>
/// <param name="options">Site runtime options supplying the concurrency limit.</param>
/// <returns>The shared gate instance.</returns>
@@ -59,7 +61,7 @@ public sealed class TriggerEvalGate : IDisposable
lock (SharedLock)
{
return _shared ??= new TriggerEvalGate(options.TriggerEvalMaxConcurrency);
return _shared ??= new TriggerEvalGate(options.TriggerEvalMaxConcurrency) { _processOwned = true };
}
}
@@ -76,6 +78,10 @@ public sealed class TriggerEvalGate : IDisposable
/// <summary>Returns a permit. Must be called exactly once per successful <see cref="WaitAsync"/>.</summary>
public void Release() => _gate.Release();
/// <inheritdoc />
public void Dispose() => _gate.Dispose();
/// <summary>Disposes a caller-created gate. The process-owned <see cref="Shared"/> gate remains available.</summary>
public void Dispose()
{
if (!_processOwned)
_gate.Dispose();
}
}
@@ -61,6 +61,10 @@ public class SharedScriptService
if (syntaxError != null)
return Result<SharedScript>.Failure(syntaxError);
var compilation = new Validation.ScriptCompiler().TryCompile(code, name);
if (compilation.IsFailure)
return Result<SharedScript>.Failure(compilation.Error);
var script = new SharedScript(name, code)
{
ParameterDefinitions = parameterDefinitions,
@@ -110,6 +114,10 @@ public class SharedScriptService
if (syntaxError != null)
return Result<SharedScript>.Failure(syntaxError);
var compilation = new Validation.ScriptCompiler().TryCompile(code, script.Name);
if (compilation.IsFailure)
return Result<SharedScript>.Failure(compilation.Error);
script.Code = code;
script.ParameterDefinitions = parameterDefinitions;
script.ReturnDefinition = returnDefinition;
@@ -171,7 +179,7 @@ public class SharedScriptService
/// comment-aware (see <see cref="Validation.CSharpDelimiterScanner"/>) so a
/// delimiter inside a regular/verbatim/interpolated/raw string literal, a
/// char literal, or a comment does not produce a false syntax error.
/// Full Roslyn compilation would be added in a later phase when the scripting sandbox is available.
/// Create and update additionally apply the cached Roslyn compile and trust gate.
/// </summary>
/// <param name="code">The C# code to validate.</param>
/// <returns>An error message if validation fails; null if valid.</returns>