From 7e61552d822add876dedf95b7269e2ed6053f926 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 5 Sep 2026 23:52:50 -0400 Subject: [PATCH] feat(wire): attribute updates carry source_execution_id/trace; PullTraceEvents RPC + site read-by-execution-ids MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flow Debugger (design 2026-09-05 §5.3/§5.4): AttributeValueUpdate fields 7/8 stamped by StreamRelayActor and parsed by SiteStreamGrpcClient (older frames => null/false). New PullTraceEvents unary RPC on SiteStreamService (same PSK-gated prefix as PullAuditEvents): a point read of the site audit SQLite by ExecutionId OR ParentExecutionId (ISiteAuditQueue.ReadByExecutionIdsAsync, json_extract on the codec's camelCase keys, new IX_audit_event_occurred index, 64-id / 2000-row caps) that moves no cursor and flips no ForwardState. Central IPullTraceEventsClient/GrpcPullTraceEventsClient mirror the audit pull client (endpoint resolution, NodeB failover, fault => empty) and register beside it. Claude-Session: https://claude.ai/code/session_01GSLkkHKz9JrAtmJjmQK3cP --- .../Central/GrpcPullTraceEventsClient.cs | 222 ++++++ .../Central/IPullTraceEventsClient.cs | 27 + .../ServiceCollectionExtensions.cs | 20 + .../Site/SqliteAuditWriter.cs | 71 ++ .../Interfaces/Services/ISiteAuditQueue.cs | 10 + .../Actors/StreamRelayActor.cs | 6 +- .../Grpc/SiteStreamGrpcClient.cs | 6 +- .../Grpc/SiteStreamGrpcServer.cs | 71 ++ .../Protos/sitestream.proto | 23 + .../SiteStreamGrpc/Sitestream.cs | 710 ++++++++++++++++-- .../SiteStreamGrpc/SitestreamGrpc.cs | 84 +++ .../Site/SiteAuditRetentionServiceTests.cs | 4 + .../SqliteAuditWriterReadByExecutionTests.cs | 83 ++ .../SiteStreamPullTraceEventsTests.cs | 125 +++ .../StreamRelayTraceStampTests.cs | 76 ++ 15 files changed, 1458 insertions(+), 80 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs create mode 100644 src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullTraceEventsClient.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterReadByExecutionTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullTraceEventsTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/StreamRelayTraceStampTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs new file mode 100644 index 00000000..a7613bcf --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs @@ -0,0 +1,222 @@ +using System.Collections.Concurrent; +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Production over the site-hosted PullTraceEvents +/// unary RPC (Flow Debugger design 2026-09-05 §5.4). Mirrors +/// exactly: endpoint resolution through +/// on every call, one NodeB failover dial on a transport fault, +/// every fault collapsed to an empty list (the debugger renders "no site rows yet" and the +/// value-change poll simply tries again), and an invoker seam for tests. +/// +public sealed class GrpcPullTraceEventsClient : IPullTraceEventsClient +{ + /// Hard cap on ids per call — matches the site's SqliteAuditWriter.MaxExecutionIdsPerRead. + public const int MaxExecutionIds = 64; + + private readonly ISiteEnumerator _sites; + private readonly IPullTraceEventsInvoker _invoker; + private readonly ILogger _logger; + + /// Creates the client over the given site enumerator and unary-call invoker. + public GrpcPullTraceEventsClient( + ISiteEnumerator sites, + IPullTraceEventsInvoker invoker, + ILogger logger) + { + _sites = sites ?? throw new ArgumentNullException(nameof(sites)); + _invoker = invoker ?? throw new ArgumentNullException(nameof(invoker)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + } + + /// + public async Task> PullAsync( + string siteId, + IReadOnlyCollection executionIds, + DateTime sinceUtc, + int maxEvents, + CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(executionIds); + if (executionIds.Count == 0 || maxEvents <= 0) + { + return Array.Empty(); + } + if (executionIds.Count > MaxExecutionIds) + { + throw new ArgumentOutOfRangeException(nameof(executionIds), $"at most {MaxExecutionIds} execution ids per call."); + } + + var (endpoint, fallback) = await ResolveEndpointsAsync(siteId, ct).ConfigureAwait(false); + if (endpoint is null) + { + _logger.LogDebug("PullTraceEvents skipped: no gRPC endpoint registered for site {SiteId}.", siteId); + return Array.Empty(); + } + + var request = new PullTraceEventsRequest + { + SinceUtc = Timestamp.FromDateTime(EnsureUtc(sinceUtc)), + MaxEvents = maxEvents, + }; + foreach (var id in executionIds) + { + request.ExecutionIds.Add(id.ToString("D")); + } + + var (reply, transportFault) = await TryInvokeAsync(endpoint, request, siteId, ct).ConfigureAwait(false); + + if (reply is null && transportFault + && !string.IsNullOrWhiteSpace(fallback) + && !string.Equals(fallback, endpoint, StringComparison.Ordinal) + && !ct.IsCancellationRequested) + { + _logger.LogInformation( + "PullTraceEvents failing over from primary {Primary} to NodeB {Fallback} for site {SiteId}.", + endpoint, fallback, siteId); + (reply, _) = await TryInvokeAsync(fallback, request, siteId, ct).ConfigureAwait(false); + } + + if (reply is null) + { + return Array.Empty(); + } + + return reply.Events + .Select(AuditEventDtoMapper.FromDto) + .OrderBy(e => e.OccurredAtUtc) + .ThenBy(e => e.EventId.ToString(), StringComparer.Ordinal) + .ToList(); + } + + private async Task<(PullTraceEventsResponse? Reply, bool TransportFault)> TryInvokeAsync( + string endpoint, PullTraceEventsRequest request, string siteId, CancellationToken ct) + { + try + { + var reply = await _invoker.InvokeAsync(siteId, endpoint, request, ct).ConfigureAwait(false); + return (reply, false); + } + catch (RpcException ex) when (ex.StatusCode is StatusCode.Unavailable or StatusCode.DeadlineExceeded or StatusCode.Cancelled) + { + _logger.LogDebug(ex, "PullTraceEvents tolerable transport fault for site {SiteId} ({Endpoint}): {Status}.", + siteId, endpoint, ex.StatusCode); + return (null, true); + } + catch (Exception ex) when (ex is HttpRequestException or System.Net.Sockets.SocketException) + { + _logger.LogDebug(ex, "PullTraceEvents connection-layer fault for site {SiteId} ({Endpoint}).", siteId, endpoint); + return (null, true); + } + catch (OperationCanceledException) + { + return (null, true); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "PullTraceEvents unexpected fault for site {SiteId} ({Endpoint}). Returning empty.", siteId, endpoint); + return (null, false); + } + } + + private async Task<(string? Primary, string? Fallback)> ResolveEndpointsAsync(string siteId, CancellationToken ct) + { + var sites = await _sites.EnumerateAsync(ct).ConfigureAwait(false); + foreach (var site in sites) + { + if (string.Equals(site.SiteId, siteId, StringComparison.Ordinal) && !string.IsNullOrWhiteSpace(site.GrpcEndpoint)) + { + return (site.GrpcEndpoint, site.FallbackGrpcEndpoint); + } + } + return (null, null); + } + + private static DateTime EnsureUtc(DateTime value) => + value.Kind == DateTimeKind.Utc ? value : DateTime.SpecifyKind(value, DateTimeKind.Utc); + + /// + /// Seam over the PullTraceEvents unary gRPC call. Production binds + /// ; tests inject a fake. + /// + public interface IPullTraceEventsInvoker + { + /// Issues the RPC against with the site's preshared key. + Task InvokeAsync( + string siteId, string endpoint, PullTraceEventsRequest request, CancellationToken ct); + } +} + +/// +/// Production : one cached +/// per (site, endpoint), keepalive from , +/// per-site PSK call credentials — the same shape as . +/// +public sealed class GrpcPullTraceEventsInvoker : GrpcPullTraceEventsClient.IPullTraceEventsInvoker, IDisposable +{ + private readonly ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel> _channels = new(); + private readonly CommunicationOptions _options; + private readonly ISitePskProvider? _pskProvider; + + /// Creates the invoker with the given options and optional per-site PSK provider. + public GrpcPullTraceEventsInvoker(CommunicationOptions options, ISitePskProvider? pskProvider) + { + _options = options ?? throw new ArgumentNullException(nameof(options)); + _pskProvider = pskProvider; + } + + /// + public async Task InvokeAsync( + string siteId, string endpoint, PullTraceEventsRequest request, CancellationToken ct) + { + var channel = GetOrCreateChannel(siteId, endpoint); + var client = new SiteStreamService.SiteStreamServiceClient(channel); + using var call = client.PullTraceEventsAsync(request, cancellationToken: ct); + return await call.ResponseAsync.ConfigureAwait(false); + } + + private GrpcChannel GetOrCreateChannel(string siteId, string endpoint) + { + var key = (siteId, endpoint); + if (!_channels.TryGetValue(key, out var channel)) + { + var created = CreateChannel(siteId, endpoint); + channel = _channels.GetOrAdd(key, created); + if (!ReferenceEquals(channel, created)) + { + created.Dispose(); + } + } + return channel; + } + + // Keyed by (site, endpoint): call credentials are per-site and bound to the channel. + private GrpcChannel CreateChannel(string siteId, string endpoint) => + GrpcChannel.ForAddress(endpoint, new GrpcChannelOptions + { + HttpHandler = new SocketsHttpHandler + { + KeepAlivePingDelay = _options.GrpcKeepAlivePingDelay, + KeepAlivePingTimeout = _options.GrpcKeepAlivePingTimeout, + KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always, + }, + }.WithSiteCredentials(_pskProvider, siteId)); + + /// + public void Dispose() + { + foreach (var channel in _channels.Values) + { + channel.Dispose(); + } + _channels.Clear(); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullTraceEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullTraceEventsClient.cs new file mode 100644 index 00000000..d78f227a --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullTraceEventsClient.cs @@ -0,0 +1,27 @@ +using ZB.MOM.WW.Audit; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Flow Debugger (design 2026-09-05 §5.4): central-side client for the site-hosted +/// PullTraceEvents unary RPC — a point read of a site's ScriptTrace rows for a +/// set of execution ids (by ExecutionId or ParentExecutionId). NOT a +/// reconciliation path: it moves no cursor and flips no forward state. Best-effort like the +/// audit pull: a transport fault collapses to an empty list after one NodeB failover dial. +/// +public interface IPullTraceEventsClient +{ + /// Pulls the site's trace rows for (max 64) since . + /// The site to pull from (resolved to its gRPC endpoint + preshared key). + /// Execution ids to match on ExecutionId or ParentExecutionId. + /// Lower bound on OccurredAtUtc (the run's start is the natural choice). + /// Row cap (the site caps at 2000). + /// Cancellation token. + /// The matching rows ordered (OccurredAtUtc, EventId); empty on fault or unknown site. + Task> PullAsync( + string siteId, + IReadOnlyCollection executionIds, + DateTime sinceUtc, + int maxEvents, + CancellationToken ct); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs index 3446bc1f..39f7b107 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs @@ -551,6 +551,26 @@ public static class ServiceCollectionExtensions sp.GetRequiredService(), sp.GetRequiredService>())); + // Flow Debugger (design 2026-09-05 §5.4): PullTraceEvents point-read client — the + // value-change entry's site-row fetch. Same invoker/channel/PSK shape as the audit + // pull client above; singleton because the invoker owns the channel cache. + services.TryAddSingleton(sp => + { + var options = sp + .GetService>(); + var psk = sp.GetService(); + return options is null + ? new GrpcPullTraceEventsInvoker(new ZB.MOM.WW.ScadaBridge.Communication.CommunicationOptions(), psk) + : new GrpcPullTraceEventsInvoker(options.Value, psk); + }); + services.TryAddSingleton( + sp => sp.GetRequiredService()); + services.TryAddSingleton(sp => new GrpcPullTraceEventsClient( + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService>())); + // Site Call Audit reconciliation pull client — central-only, the // sibling of the audit pull client above. Lives here (not in the // SiteCallAudit project) so it can reuse the central-only diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs index 775729ac..3db9c7ba 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs @@ -265,6 +265,12 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable -- four reads + the backlog COUNT/MIN. CREATE INDEX IF NOT EXISTS IX_fwd ON audit_forward_state (ForwardState, IsCachedKind, OccurredAtUtc); + + -- Flow Debugger (design 2026-09-05 §5.4): PullTraceEvents reads audit_event by + -- time window + json_extract on ExecutionId/ParentExecutionId; this bounds the scan + -- to the window. Additive and idempotent. + CREATE INDEX IF NOT EXISTS IX_audit_event_occurred + ON audit_event (OccurredAtUtc); """; cmd.ExecuteNonQuery(); } @@ -848,6 +854,71 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable } } + /// Flow Debugger PullTraceEvents: hard cap on ids per read (design §5.4). + public const int MaxExecutionIdsPerRead = 64; + + /// + public Task> ReadByExecutionIdsAsync( + IReadOnlyCollection executionIds, DateTime sinceUtc, int limit, CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(executionIds); + if (limit <= 0) + { + throw new ArgumentOutOfRangeException(nameof(limit), "limit must be > 0."); + } + if (executionIds.Count == 0) + { + return Task.FromResult>(Array.Empty()); + } + if (executionIds.Count > MaxExecutionIdsPerRead) + { + throw new ArgumentOutOfRangeException(nameof(executionIds), + $"at most {MaxExecutionIdsPerRead} execution ids per read."); + } + + // Flow Debugger (design 2026-09-05 §5.4): a POINT read independent of the forwarding + // sidecar — a traced row is served whether Pending, Forwarded or Reconciled, and + // serving it moves no cursor. DetailsJson is the codec's camelCase JSON, so the paths + // are $.executionId / $.parentExecutionId; ids are stored in "D" form. + lock (_readLock) + { + ObjectDisposedException.ThrowIf(_disposed, this); + + using var cmd = _readConnection.CreateCommand(); + var names = new List(executionIds.Count); + var i = 0; + foreach (var id in executionIds) + { + var name = $"$e{i++}"; + names.Add(name); + cmd.Parameters.AddWithValue(name, id.ToString("D")); + } + 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.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString( + "o", System.Globalization.CultureInfo.InvariantCulture)); + cmd.Parameters.AddWithValue("$limit", limit); + + var rows = new List(Math.Min(limit, 256)); + using var reader = cmd.ExecuteReader(); + while (reader.Read()) + { + rows.Add(MapRow(reader)); + } + return Task.FromResult>(rows); + } + } + /// /// Highest audit_forward_state.rowid this instance has ever SERVED from /// — i.e. the insertion-order high-water mark of diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs index be3fc2ff..7288a7e9 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs @@ -223,4 +223,14 @@ public interface ISiteAuditQueue /// Cancellation token. /// A task that resolves to the number of canonical rows deleted. Task PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default); + + /// + /// Flow Debugger (design 2026-09-05 §5.4): point read of the rows whose + /// ExecutionId OR ParentExecutionId is one of + /// and whose OccurredAtUtc is at or after , ordered + /// (OccurredAtUtc, EventId), at most rows. Independent of + /// forwarding state — moves no cursor and flips nothing. At most 64 ids per call. + /// + Task> ReadByExecutionIdsAsync( + IReadOnlyCollection executionIds, DateTime sinceUtc, int limit, CancellationToken ct = default); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs index 6046e27e..d5bcf1cd 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs @@ -47,7 +47,11 @@ public class StreamRelayActor : ReceiveActor AttributeName = msg.AttributeName, Value = AttributeValueCodec.Encode(msg.Value) ?? string.Empty, Quality = MapQuality(msg.Quality), - Timestamp = Timestamp.FromDateTimeOffset(msg.Timestamp) + Timestamp = Timestamp.FromDateTimeOffset(msg.Timestamp), + // Flow Debugger (design 2026-09-05 §5.3/§5.4): additive fields 7/8 — empty / + // false for every change that did not come from a traced write. + SourceExecutionId = msg.SourceExecutionId?.ToString("D") ?? string.Empty, + Trace = msg.Trace } }; diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs index 42c5539e..fe163f9d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs @@ -557,7 +557,11 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable evt.AttributeChanged.AttributeName, evt.AttributeChanged.Value, MapQuality(evt.AttributeChanged.Quality), - evt.AttributeChanged.Timestamp.ToDateTimeOffset()), + evt.AttributeChanged.Timestamp.ToDateTimeOffset(), + // Flow Debugger (design §5.4): fields 7/8 — an older site leaves them at the + // proto3 defaults, which map to null / false. + SourceExecutionId: Guid.TryParse(evt.AttributeChanged.SourceExecutionId, out var traceSource) ? traceSource : null, + Trace: evt.AttributeChanged.Trace), SiteStreamEvent.EventOneofCase.AlarmChanged => new AlarmStateChanged( evt.AlarmChanged.InstanceUniqueName, evt.AlarmChanged.AlarmName, diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs index dd15e604..352c39f0 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs @@ -641,6 +641,77 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase return ack; } + /// Flow Debugger PullTraceEvents: server-side row cap (design 2026-09-05 §5.4). + public const int MaxTraceEventsPerPull = 2000; + + /// Flow Debugger PullTraceEvents: server-side id cap, matching the site queue's. + public const int MaxTraceExecutionIdsPerPull = 64; + + /// + /// Flow Debugger (design 2026-09-05 §5.4): point read of this site's audit rows whose + /// ExecutionId or ParentExecutionId is one of the requested ids. NOT a + /// reconciliation path — moves no cursor, flips no ForwardState. Gated by the same + /// ControlPlaneAuthInterceptor prefix as . + /// + public override async Task PullTraceEvents( + PullTraceEventsRequest request, + ServerCallContext context) + { + var queue = _siteAuditQueue; + if (queue is null) + { + _logger.LogWarning( + "PullTraceEvents invoked before SetSiteAuditQueue was called; returning empty response."); + return new PullTraceEventsResponse(); + } + + if (request.MaxEvents <= 0 || request.MaxEvents > MaxTraceEventsPerPull) + { + throw new RpcException(new GrpcStatus( + StatusCode.InvalidArgument, $"max_events must be 1..{MaxTraceEventsPerPull}")); + } + if (request.ExecutionIds.Count == 0 || request.ExecutionIds.Count > MaxTraceExecutionIdsPerPull) + { + throw new RpcException(new GrpcStatus( + StatusCode.InvalidArgument, $"execution_ids must hold 1..{MaxTraceExecutionIdsPerPull} ids")); + } + + var ids = new List(request.ExecutionIds.Count); + foreach (var raw in request.ExecutionIds) + { + if (!Guid.TryParse(raw, out var id)) + { + throw new RpcException(new GrpcStatus( + StatusCode.InvalidArgument, $"execution_ids entry '{raw}' is not a GUID")); + } + ids.Add(id); + } + + var since = request.SinceUtc is not null + ? DateTime.SpecifyKind(request.SinceUtc.ToDateTime(), DateTimeKind.Utc) + : DateTime.MinValue; + + IReadOnlyList events; + try + { + events = await queue.ReadByExecutionIdsAsync(ids, since, request.MaxEvents, context.CancellationToken); + } + catch (Exception ex) + { + _logger.LogError(ex, + "ReadByExecutionIdsAsync failed for {Count} ids since={Since}; returning empty response.", + ids.Count, since); + return new PullTraceEventsResponse(); + } + + var response = new PullTraceEventsResponse(); + foreach (var evt in events) + { + response.Events.Add(AuditEventDtoMapper.ToDto(evt)); + } + return response; + } + /// public override async Task PullAuditEvents( PullAuditEventsRequest request, diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto index 831f9e89..e8829bf9 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto @@ -14,6 +14,10 @@ service SiteStreamService { rpc IngestAuditEvents(AuditEventBatch) returns (IngestAck); rpc IngestCachedTelemetry(CachedTelemetryBatch) returns (IngestAck); rpc PullAuditEvents(PullAuditEventsRequest) returns (PullAuditEventsResponse); + // Flow Debugger (design 2026-09-05 §5.4): point read of a site's ScriptTrace rows for a + // set of execution ids (by ExecutionId OR ParentExecutionId). NOT a reconciliation path — + // moves no cursor, flips no ForwardState. Same PSK interceptor as PullAuditEvents. + rpc PullTraceEvents(PullTraceEventsRequest) returns (PullTraceEventsResponse); rpc PullSiteCalls(PullSiteCallsRequest) returns (PullSiteCallsResponse); } @@ -95,6 +99,11 @@ message AttributeValueUpdate { string value = 4; Quality quality = 5; google.protobuf.Timestamp timestamp = 6; + // 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. + string source_execution_id = 7; + bool trace = 8; } message AlarmStateUpdate { @@ -218,6 +227,20 @@ message PullAuditEventsResponse { bool more_available = 2; } +// Flow Debugger (design 2026-09-05 §5.4): PullTraceEvents request/response. The site +// returns audit rows whose ExecutionId or ParentExecutionId is in execution_ids and whose +// OccurredAtUtc >= since_utc, ordered (OccurredAtUtc, EventId), at most max_events (server +// caps at 2000). Ids capped at 64 per call. +message PullTraceEventsRequest { + repeated string execution_ids = 1; // "D" GUID form + google.protobuf.Timestamp since_utc = 2; + int32 max_events = 3; +} + +message PullTraceEventsResponse { + repeated AuditEventDto events = 1; +} + // Site Call Audit (#22) reconciliation pull: central→site request for any // site-local operation-tracking rows whose UpdatedAtUtc >= since_utc — the // self-heal feed that backfills the eventually-consistent central SiteCalls diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs index ab2d8dfc..ac347c0c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs @@ -36,85 +36,92 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { "ZBgDIAEoCzIcLnNpdGVzdHJlYW0uQWxhcm1TdGF0ZVVwZGF0ZUgAEjEKBWJh", "dGNoGAQgASgLMiAuc2l0ZXN0cmVhbS5TaXRlU3RyZWFtRXZlbnRCYXRjaEgA", "QgcKBWV2ZW50IkMKFFNpdGVTdHJlYW1FdmVudEJhdGNoEisKBmV2ZW50cxgB", - "IAMoCzIbLnNpdGVzdHJlYW0uU2l0ZVN0cmVhbUV2ZW50IsgBChRBdHRyaWJ1", + "IAMoCzIbLnNpdGVzdHJlYW0uU2l0ZVN0cmVhbUV2ZW50IvQBChRBdHRyaWJ1", "dGVWYWx1ZVVwZGF0ZRIcChRpbnN0YW5jZV91bmlxdWVfbmFtZRgBIAEoCRIW", "Cg5hdHRyaWJ1dGVfcGF0aBgCIAEoCRIWCg5hdHRyaWJ1dGVfbmFtZRgDIAEo", "CRINCgV2YWx1ZRgEIAEoCRIkCgdxdWFsaXR5GAUgASgOMhMuc2l0ZXN0cmVh", "bS5RdWFsaXR5Ei0KCXRpbWVzdGFtcBgGIAEoCzIaLmdvb2dsZS5wcm90b2J1", - "Zi5UaW1lc3RhbXAirwUKEEFsYXJtU3RhdGVVcGRhdGUSHAoUaW5zdGFuY2Vf", - "dW5pcXVlX25hbWUYASABKAkSEgoKYWxhcm1fbmFtZRgCIAEoCRIpCgVzdGF0", - "ZRgDIAEoDjIaLnNpdGVzdHJlYW0uQWxhcm1TdGF0ZUVudW0SEAoIcHJpb3Jp", - "dHkYBCABKAUSLQoJdGltZXN0YW1wGAUgASgLMhouZ29vZ2xlLnByb3RvYnVm", - "LlRpbWVzdGFtcBIpCgVsZXZlbBgGIAEoDjIaLnNpdGVzdHJlYW0uQWxhcm1M", - "ZXZlbEVudW0SDwoHbWVzc2FnZRgHIAEoCRIMCgRraW5kGAggASgJEg4KBmFj", - "dGl2ZRgJIAEoCBIUCgxhY2tub3dsZWRnZWQYCiABKAgSEQoJY29uZmlybWVk", - "GAsgASgIEhQKDHNoZWx2ZV9zdGF0ZRgMIAEoCRISCgpzdXBwcmVzc2VkGA0g", - "ASgIEhgKEHNvdXJjZV9yZWZlcmVuY2UYDiABKAkSFwoPYWxhcm1fdHlwZV9u", - "YW1lGA8gASgJEhAKCGNhdGVnb3J5GBAgASgJEhUKDW9wZXJhdG9yX3VzZXIY", - "ESABKAkSGAoQb3BlcmF0b3JfY29tbWVudBgSIAEoCRI3ChNvcmlnaW5hbF9y", - "YWlzZV90aW1lGBMgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIV", - "Cg1jdXJyZW50X3ZhbHVlGBQgASgJEhMKC2xpbWl0X3ZhbHVlGBUgASgJEiQK", - "HG5hdGl2ZV9zb3VyY2VfY2Fub25pY2FsX25hbWUYFiABKAkSIQoZaXNfY29u", - "ZmlndXJlZF9wbGFjZWhvbGRlchgXIAEoCBIsCghhY2tfdGltZRgYIAEoCzIa", - "Lmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXAivQQKDUF1ZGl0RXZlbnREdG8S", - "EAoIZXZlbnRfaWQYASABKAkSMwoPb2NjdXJyZWRfYXRfdXRjGAIgASgLMhou", - "Z29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIPCgdjaGFubmVsGAMgASgJEgwK", - "BGtpbmQYBCABKAkSFgoOY29ycmVsYXRpb25faWQYBSABKAkSFgoOc291cmNl", - "X3NpdGVfaWQYBiABKAkSGgoSc291cmNlX2luc3RhbmNlX2lkGAcgASgJEhUK", - "DXNvdXJjZV9zY3JpcHQYCCABKAkSDQoFYWN0b3IYCSABKAkSDgoGdGFyZ2V0", - "GAogASgJEg4KBnN0YXR1cxgLIAEoCRIwCgtodHRwX3N0YXR1cxgMIAEoCzIb", - "Lmdvb2dsZS5wcm90b2J1Zi5JbnQzMlZhbHVlEjAKC2R1cmF0aW9uX21zGA0g", - "ASgLMhsuZ29vZ2xlLnByb3RvYnVmLkludDMyVmFsdWUSFQoNZXJyb3JfbWVz", - "c2FnZRgOIAEoCRIUCgxlcnJvcl9kZXRhaWwYDyABKAkSFwoPcmVxdWVzdF9z", - "dW1tYXJ5GBAgASgJEhgKEHJlc3BvbnNlX3N1bW1hcnkYESABKAkSGQoRcGF5", - "bG9hZF90cnVuY2F0ZWQYEiABKAgSDQoFZXh0cmEYEyABKAkSFAoMZXhlY3V0", - "aW9uX2lkGBQgASgJEhsKE3BhcmVudF9leGVjdXRpb25faWQYFSABKAkSEwoL", - "c291cmNlX25vZGUYFiABKAkiPAoPQXVkaXRFdmVudEJhdGNoEikKBmV2ZW50", - "cxgBIAMoCzIZLnNpdGVzdHJlYW0uQXVkaXRFdmVudER0byInCglJbmdlc3RB", - "Y2sSGgoSYWNjZXB0ZWRfZXZlbnRfaWRzGAEgAygJIokDChZTaXRlQ2FsbE9w", - "ZXJhdGlvbmFsRHRvEhwKFHRyYWNrZWRfb3BlcmF0aW9uX2lkGAEgASgJEg8K", - "B2NoYW5uZWwYAiABKAkSDgoGdGFyZ2V0GAMgASgJEhMKC3NvdXJjZV9zaXRl", - "GAQgASgJEg4KBnN0YXR1cxgFIAEoCRITCgtyZXRyeV9jb3VudBgGIAEoBRIS", - "CgpsYXN0X2Vycm9yGAcgASgJEjAKC2h0dHBfc3RhdHVzGAggASgLMhsuZ29v", - "Z2xlLnByb3RvYnVmLkludDMyVmFsdWUSMgoOY3JlYXRlZF9hdF91dGMYCSAB", - "KAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEjIKDnVwZGF0ZWRfYXRf", - "dXRjGAogASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIzCg90ZXJt", - "aW5hbF9hdF91dGMYCyABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1w", - "EhMKC3NvdXJjZV9ub2RlGAwgASgJIoABChVDYWNoZWRUZWxlbWV0cnlQYWNr", - "ZXQSLgoLYXVkaXRfZXZlbnQYASABKAsyGS5zaXRlc3RyZWFtLkF1ZGl0RXZl", - "bnREdG8SNwoLb3BlcmF0aW9uYWwYAiABKAsyIi5zaXRlc3RyZWFtLlNpdGVD", - "YWxsT3BlcmF0aW9uYWxEdG8iSgoUQ2FjaGVkVGVsZW1ldHJ5QmF0Y2gSMgoH", - "cGFja2V0cxgBIAMoCzIhLnNpdGVzdHJlYW0uQ2FjaGVkVGVsZW1ldHJ5UGFj", - "a2V0Im0KFlB1bGxBdWRpdEV2ZW50c1JlcXVlc3QSLQoJc2luY2VfdXRjGAEg", - "ASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBISCgpiYXRjaF9zaXpl", - "GAIgASgFEhAKCGFmdGVyX2lkGAMgASgJIlwKF1B1bGxBdWRpdEV2ZW50c1Jl", - "c3BvbnNlEikKBmV2ZW50cxgBIAMoCzIZLnNpdGVzdHJlYW0uQXVkaXRFdmVu", - "dER0bxIWCg5tb3JlX2F2YWlsYWJsZRgCIAEoCCJrChRQdWxsU2l0ZUNhbGxz", - "UmVxdWVzdBItCglzaW5jZV91dGMYASABKAsyGi5nb29nbGUucHJvdG9idWYu", - "VGltZXN0YW1wEhIKCmJhdGNoX3NpemUYAiABKAUSEAoIYWZ0ZXJfaWQYAyAB", - "KAkiaQoVUHVsbFNpdGVDYWxsc1Jlc3BvbnNlEjgKDG9wZXJhdGlvbmFscxgB", - "IAMoCzIiLnNpdGVzdHJlYW0uU2l0ZUNhbGxPcGVyYXRpb25hbER0bxIWCg5t", - "b3JlX2F2YWlsYWJsZRgCIAEoCCpcCgdRdWFsaXR5EhcKE1FVQUxJVFlfVU5T", - "UEVDSUZJRUQQABIQCgxRVUFMSVRZX0dPT0QQARIVChFRVUFMSVRZX1VOQ0VS", - "VEFJThACEg8KC1FVQUxJVFlfQkFEEAMqXQoOQWxhcm1TdGF0ZUVudW0SGwoX", - "QUxBUk1fU1RBVEVfVU5TUEVDSUZJRUQQABIWChJBTEFSTV9TVEFURV9OT1JN", - "QUwQARIWChJBTEFSTV9TVEFURV9BQ1RJVkUQAiqFAQoOQWxhcm1MZXZlbEVu", - "dW0SFAoQQUxBUk1fTEVWRUxfTk9ORRAAEhMKD0FMQVJNX0xFVkVMX0xPVxAB", - "EhcKE0FMQVJNX0xFVkVMX0xPV19MT1cQAhIUChBBTEFSTV9MRVZFTF9ISUdI", - "EAMSGQoVQUxBUk1fTEVWRUxfSElHSF9ISUdIEAQyhgQKEVNpdGVTdHJlYW1T", - "ZXJ2aWNlElUKEVN1YnNjcmliZUluc3RhbmNlEiEuc2l0ZXN0cmVhbS5JbnN0", - "YW5jZVN0cmVhbVJlcXVlc3QaGy5zaXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVu", - "dDABEk0KDVN1YnNjcmliZVNpdGUSHS5zaXRlc3RyZWFtLlNpdGVTdHJlYW1S", - "ZXF1ZXN0Ghsuc2l0ZXN0cmVhbS5TaXRlU3RyZWFtRXZlbnQwARJHChFJbmdl", - "c3RBdWRpdEV2ZW50cxIbLnNpdGVzdHJlYW0uQXVkaXRFdmVudEJhdGNoGhUu", - "c2l0ZXN0cmVhbS5Jbmdlc3RBY2sSUAoVSW5nZXN0Q2FjaGVkVGVsZW1ldHJ5", - "EiAuc2l0ZXN0cmVhbS5DYWNoZWRUZWxlbWV0cnlCYXRjaBoVLnNpdGVzdHJl", - "YW0uSW5nZXN0QWNrEloKD1B1bGxBdWRpdEV2ZW50cxIiLnNpdGVzdHJlYW0u", - "UHVsbEF1ZGl0RXZlbnRzUmVxdWVzdBojLnNpdGVzdHJlYW0uUHVsbEF1ZGl0", - "RXZlbnRzUmVzcG9uc2USVAoNUHVsbFNpdGVDYWxscxIgLnNpdGVzdHJlYW0u", - "UHVsbFNpdGVDYWxsc1JlcXVlc3QaIS5zaXRlc3RyZWFtLlB1bGxTaXRlQ2Fs", - "bHNSZXNwb25zZUIrqgIoWkIuTU9NLldXLlNjYWRhQnJpZGdlLkNvbW11bmlj", - "YXRpb24uR3JwY2IGcHJvdG8z")); + "Zi5UaW1lc3RhbXASGwoTc291cmNlX2V4ZWN1dGlvbl9pZBgHIAEoCRINCgV0", + "cmFjZRgIIAEoCCKvBQoQQWxhcm1TdGF0ZVVwZGF0ZRIcChRpbnN0YW5jZV91", + "bmlxdWVfbmFtZRgBIAEoCRISCgphbGFybV9uYW1lGAIgASgJEikKBXN0YXRl", + "GAMgASgOMhouc2l0ZXN0cmVhbS5BbGFybVN0YXRlRW51bRIQCghwcmlvcml0", + "eRgEIAEoBRItCgl0aW1lc3RhbXAYBSABKAsyGi5nb29nbGUucHJvdG9idWYu", + "VGltZXN0YW1wEikKBWxldmVsGAYgASgOMhouc2l0ZXN0cmVhbS5BbGFybUxl", + "dmVsRW51bRIPCgdtZXNzYWdlGAcgASgJEgwKBGtpbmQYCCABKAkSDgoGYWN0", + "aXZlGAkgASgIEhQKDGFja25vd2xlZGdlZBgKIAEoCBIRCgljb25maXJtZWQY", + "CyABKAgSFAoMc2hlbHZlX3N0YXRlGAwgASgJEhIKCnN1cHByZXNzZWQYDSAB", + "KAgSGAoQc291cmNlX3JlZmVyZW5jZRgOIAEoCRIXCg9hbGFybV90eXBlX25h", + "bWUYDyABKAkSEAoIY2F0ZWdvcnkYECABKAkSFQoNb3BlcmF0b3JfdXNlchgR", + "IAEoCRIYChBvcGVyYXRvcl9jb21tZW50GBIgASgJEjcKE29yaWdpbmFsX3Jh", + "aXNlX3RpbWUYEyABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhUK", + "DWN1cnJlbnRfdmFsdWUYFCABKAkSEwoLbGltaXRfdmFsdWUYFSABKAkSJAoc", + "bmF0aXZlX3NvdXJjZV9jYW5vbmljYWxfbmFtZRgWIAEoCRIhChlpc19jb25m", + "aWd1cmVkX3BsYWNlaG9sZGVyGBcgASgIEiwKCGFja190aW1lGBggASgLMhou", + "Z29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcCK9BAoNQXVkaXRFdmVudER0bxIQ", + "CghldmVudF9pZBgBIAEoCRIzCg9vY2N1cnJlZF9hdF91dGMYAiABKAsyGi5n", + "b29nbGUucHJvdG9idWYuVGltZXN0YW1wEg8KB2NoYW5uZWwYAyABKAkSDAoE", + "a2luZBgEIAEoCRIWCg5jb3JyZWxhdGlvbl9pZBgFIAEoCRIWCg5zb3VyY2Vf", + "c2l0ZV9pZBgGIAEoCRIaChJzb3VyY2VfaW5zdGFuY2VfaWQYByABKAkSFQoN", + "c291cmNlX3NjcmlwdBgIIAEoCRINCgVhY3RvchgJIAEoCRIOCgZ0YXJnZXQY", + "CiABKAkSDgoGc3RhdHVzGAsgASgJEjAKC2h0dHBfc3RhdHVzGAwgASgLMhsu", + "Z29vZ2xlLnByb3RvYnVmLkludDMyVmFsdWUSMAoLZHVyYXRpb25fbXMYDSAB", + "KAsyGy5nb29nbGUucHJvdG9idWYuSW50MzJWYWx1ZRIVCg1lcnJvcl9tZXNz", + "YWdlGA4gASgJEhQKDGVycm9yX2RldGFpbBgPIAEoCRIXCg9yZXF1ZXN0X3N1", + "bW1hcnkYECABKAkSGAoQcmVzcG9uc2Vfc3VtbWFyeRgRIAEoCRIZChFwYXls", + "b2FkX3RydW5jYXRlZBgSIAEoCBINCgVleHRyYRgTIAEoCRIUCgxleGVjdXRp", + "b25faWQYFCABKAkSGwoTcGFyZW50X2V4ZWN1dGlvbl9pZBgVIAEoCRITCgtz", + "b3VyY2Vfbm9kZRgWIAEoCSI8Cg9BdWRpdEV2ZW50QmF0Y2gSKQoGZXZlbnRz", + "GAEgAygLMhkuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50RHRvIicKCUluZ2VzdEFj", + "axIaChJhY2NlcHRlZF9ldmVudF9pZHMYASADKAkiiQMKFlNpdGVDYWxsT3Bl", + "cmF0aW9uYWxEdG8SHAoUdHJhY2tlZF9vcGVyYXRpb25faWQYASABKAkSDwoH", + "Y2hhbm5lbBgCIAEoCRIOCgZ0YXJnZXQYAyABKAkSEwoLc291cmNlX3NpdGUY", + "BCABKAkSDgoGc3RhdHVzGAUgASgJEhMKC3JldHJ5X2NvdW50GAYgASgFEhIK", + "Cmxhc3RfZXJyb3IYByABKAkSMAoLaHR0cF9zdGF0dXMYCCABKAsyGy5nb29n", + "bGUucHJvdG9idWYuSW50MzJWYWx1ZRIyCg5jcmVhdGVkX2F0X3V0YxgJIAEo", + "CzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXASMgoOdXBkYXRlZF9hdF91", + "dGMYCiABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEjMKD3Rlcm1p", + "bmFsX2F0X3V0YxgLIAEoCzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXAS", + "EwoLc291cmNlX25vZGUYDCABKAkigAEKFUNhY2hlZFRlbGVtZXRyeVBhY2tl", + "dBIuCgthdWRpdF9ldmVudBgBIAEoCzIZLnNpdGVzdHJlYW0uQXVkaXRFdmVu", + "dER0bxI3CgtvcGVyYXRpb25hbBgCIAEoCzIiLnNpdGVzdHJlYW0uU2l0ZUNh", + "bGxPcGVyYXRpb25hbER0byJKChRDYWNoZWRUZWxlbWV0cnlCYXRjaBIyCgdw", + "YWNrZXRzGAEgAygLMiEuc2l0ZXN0cmVhbS5DYWNoZWRUZWxlbWV0cnlQYWNr", + "ZXQibQoWUHVsbEF1ZGl0RXZlbnRzUmVxdWVzdBItCglzaW5jZV91dGMYASAB", + "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==")); 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[] { @@ -122,7 +129,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamRequest.Parser, new[]{ "CorrelationId", "BatchingSupported" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamEvent), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamEvent.Parser, new[]{ "CorrelationId", "AttributeChanged", "AlarmChanged", "Batch" }, new[]{ "Event" }, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamEventBatch), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamEventBatch.Parser, new[]{ "Events" }, null, null, null, null), - new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AttributeValueUpdate), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AttributeValueUpdate.Parser, new[]{ "InstanceUniqueName", "AttributePath", "AttributeName", "Value", "Quality", "Timestamp" }, null, null, null, null), + new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AttributeValueUpdate), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AttributeValueUpdate.Parser, new[]{ "InstanceUniqueName", "AttributePath", "AttributeName", "Value", "Quality", "Timestamp", "SourceExecutionId", "Trace" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateUpdate), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateUpdate.Parser, new[]{ "InstanceUniqueName", "AlarmName", "State", "Priority", "Timestamp", "Level", "Message", "Kind", "Active", "Acknowledged", "Confirmed", "ShelveState", "Suppressed", "SourceReference", "AlarmTypeName", "Category", "OperatorUser", "OperatorComment", "OriginalRaiseTime", "CurrentValue", "LimitValue", "NativeSourceCanonicalName", "IsConfiguredPlaceholder", "AckTime" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventDto), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventDto.Parser, new[]{ "EventId", "OccurredAtUtc", "Channel", "Kind", "CorrelationId", "SourceSiteId", "SourceInstanceId", "SourceScript", "Actor", "Target", "Status", "HttpStatus", "DurationMs", "ErrorMessage", "ErrorDetail", "RequestSummary", "ResponseSummary", "PayloadTruncated", "Extra", "ExecutionId", "ParentExecutionId", "SourceNode" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventBatch), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventBatch.Parser, new[]{ "Events" }, null, null, null, null), @@ -132,6 +139,8 @@ 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.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) })); @@ -1321,6 +1330,8 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { value_ = other.value_; quality_ = other.quality_; timestamp_ = other.timestamp_ != null ? other.timestamp_.Clone() : null; + sourceExecutionId_ = other.sourceExecutionId_; + trace_ = other.trace_; _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); } @@ -1402,6 +1413,35 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } } + /// Field number for the "source_execution_id" field. + public const int SourceExecutionIdFieldNumber = 7; + private string sourceExecutionId_ = ""; + /// + /// 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. + /// + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public string SourceExecutionId { + get { return sourceExecutionId_; } + set { + sourceExecutionId_ = pb::ProtoPreconditions.CheckNotNull(value, "value"); + } + } + + /// Field number for the "trace" field. + public const int TraceFieldNumber = 8; + private bool trace_; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public bool Trace { + get { return trace_; } + set { + trace_ = value; + } + } + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public override bool Equals(object other) { @@ -1423,6 +1463,8 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { if (Value != other.Value) return false; if (Quality != other.Quality) return false; if (!object.Equals(Timestamp, other.Timestamp)) return false; + if (SourceExecutionId != other.SourceExecutionId) return false; + if (Trace != other.Trace) return false; return Equals(_unknownFields, other._unknownFields); } @@ -1436,6 +1478,8 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { if (Value.Length != 0) hash ^= Value.GetHashCode(); if (Quality != global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.Quality.Unspecified) hash ^= Quality.GetHashCode(); if (timestamp_ != null) hash ^= Timestamp.GetHashCode(); + if (SourceExecutionId.Length != 0) hash ^= SourceExecutionId.GetHashCode(); + if (Trace != false) hash ^= Trace.GetHashCode(); if (_unknownFields != null) { hash ^= _unknownFields.GetHashCode(); } @@ -1478,6 +1522,14 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { output.WriteRawTag(50); output.WriteMessage(Timestamp); } + if (SourceExecutionId.Length != 0) { + output.WriteRawTag(58); + output.WriteString(SourceExecutionId); + } + if (Trace != false) { + output.WriteRawTag(64); + output.WriteBool(Trace); + } if (_unknownFields != null) { _unknownFields.WriteTo(output); } @@ -1512,6 +1564,14 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { output.WriteRawTag(50); output.WriteMessage(Timestamp); } + if (SourceExecutionId.Length != 0) { + output.WriteRawTag(58); + output.WriteString(SourceExecutionId); + } + if (Trace != false) { + output.WriteRawTag(64); + output.WriteBool(Trace); + } if (_unknownFields != null) { _unknownFields.WriteTo(ref output); } @@ -1540,6 +1600,12 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { if (timestamp_ != null) { size += 1 + pb::CodedOutputStream.ComputeMessageSize(Timestamp); } + if (SourceExecutionId.Length != 0) { + size += 1 + pb::CodedOutputStream.ComputeStringSize(SourceExecutionId); + } + if (Trace != false) { + size += 1 + 1; + } if (_unknownFields != null) { size += _unknownFields.CalculateSize(); } @@ -1573,6 +1639,12 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } Timestamp.MergeFrom(other.Timestamp); } + if (other.SourceExecutionId.Length != 0) { + SourceExecutionId = other.SourceExecutionId; + } + if (other.Trace != false) { + Trace = other.Trace; + } _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); } @@ -1619,6 +1691,14 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { input.ReadMessage(Timestamp); break; } + case 58: { + SourceExecutionId = input.ReadString(); + break; + } + case 64: { + Trace = input.ReadBool(); + break; + } } } #endif @@ -1665,6 +1745,14 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { input.ReadMessage(Timestamp); break; } + case 58: { + SourceExecutionId = input.ReadString(); + break; + } + case 64: { + Trace = input.ReadBool(); + break; + } } } } @@ -5813,6 +5901,472 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } + /// + /// Flow Debugger (design 2026-09-05 §5.4): PullTraceEvents request/response. The site + /// returns audit rows whose ExecutionId or ParentExecutionId is in execution_ids and whose + /// OccurredAtUtc >= since_utc, ordered (OccurredAtUtc, EventId), at most max_events (server + /// caps at 2000). Ids capped at 64 per call. + /// + [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] + public sealed partial class PullTraceEventsRequest : pb::IMessage + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + , pb::IBufferMessage + #endif + { + private static readonly pb::MessageParser _parser = new pb::MessageParser(() => new PullTraceEventsRequest()); + private pb::UnknownFieldSet _unknownFields; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pb::MessageParser Parser { get { return _parser; } } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pbr::MessageDescriptor Descriptor { + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[14]; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + pbr::MessageDescriptor pb::IMessage.Descriptor { + get { return Descriptor; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullTraceEventsRequest() { + OnConstruction(); + } + + partial void OnConstruction(); + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullTraceEventsRequest(PullTraceEventsRequest other) : this() { + executionIds_ = other.executionIds_.Clone(); + sinceUtc_ = other.sinceUtc_ != null ? other.sinceUtc_.Clone() : null; + maxEvents_ = other.maxEvents_; + _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullTraceEventsRequest Clone() { + return new PullTraceEventsRequest(this); + } + + /// Field number for the "execution_ids" field. + public const int ExecutionIdsFieldNumber = 1; + private static readonly pb::FieldCodec _repeated_executionIds_codec + = pb::FieldCodec.ForString(10); + private readonly pbc::RepeatedField executionIds_ = new pbc::RepeatedField(); + /// + /// "D" GUID form + /// + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public pbc::RepeatedField ExecutionIds { + get { return executionIds_; } + } + + /// Field number for the "since_utc" field. + public const int SinceUtcFieldNumber = 2; + private global::Google.Protobuf.WellKnownTypes.Timestamp sinceUtc_; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public global::Google.Protobuf.WellKnownTypes.Timestamp SinceUtc { + get { return sinceUtc_; } + set { + sinceUtc_ = value; + } + } + + /// Field number for the "max_events" field. + public const int MaxEventsFieldNumber = 3; + private int maxEvents_; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public int MaxEvents { + get { return maxEvents_; } + set { + maxEvents_ = value; + } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override bool Equals(object other) { + return Equals(other as PullTraceEventsRequest); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public bool Equals(PullTraceEventsRequest other) { + if (ReferenceEquals(other, null)) { + return false; + } + if (ReferenceEquals(other, this)) { + return true; + } + if(!executionIds_.Equals(other.executionIds_)) return false; + if (!object.Equals(SinceUtc, other.SinceUtc)) return false; + if (MaxEvents != other.MaxEvents) return false; + return Equals(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override int GetHashCode() { + int hash = 1; + hash ^= executionIds_.GetHashCode(); + if (sinceUtc_ != null) hash ^= SinceUtc.GetHashCode(); + if (MaxEvents != 0) hash ^= MaxEvents.GetHashCode(); + if (_unknownFields != null) { + hash ^= _unknownFields.GetHashCode(); + } + return hash; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override string ToString() { + return pb::JsonFormatter.ToDiagnosticString(this); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void WriteTo(pb::CodedOutputStream output) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + output.WriteRawMessage(this); + #else + executionIds_.WriteTo(output, _repeated_executionIds_codec); + if (sinceUtc_ != null) { + output.WriteRawTag(18); + output.WriteMessage(SinceUtc); + } + if (MaxEvents != 0) { + output.WriteRawTag(24); + output.WriteInt32(MaxEvents); + } + if (_unknownFields != null) { + _unknownFields.WriteTo(output); + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalWriteTo(ref pb::WriteContext output) { + executionIds_.WriteTo(ref output, _repeated_executionIds_codec); + if (sinceUtc_ != null) { + output.WriteRawTag(18); + output.WriteMessage(SinceUtc); + } + if (MaxEvents != 0) { + output.WriteRawTag(24); + output.WriteInt32(MaxEvents); + } + if (_unknownFields != null) { + _unknownFields.WriteTo(ref output); + } + } + #endif + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public int CalculateSize() { + int size = 0; + size += executionIds_.CalculateSize(_repeated_executionIds_codec); + if (sinceUtc_ != null) { + size += 1 + pb::CodedOutputStream.ComputeMessageSize(SinceUtc); + } + if (MaxEvents != 0) { + size += 1 + pb::CodedOutputStream.ComputeInt32Size(MaxEvents); + } + if (_unknownFields != null) { + size += _unknownFields.CalculateSize(); + } + return size; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(PullTraceEventsRequest other) { + if (other == null) { + return; + } + executionIds_.Add(other.executionIds_); + if (other.sinceUtc_ != null) { + if (sinceUtc_ == null) { + SinceUtc = new global::Google.Protobuf.WellKnownTypes.Timestamp(); + } + SinceUtc.MergeFrom(other.SinceUtc); + } + if (other.MaxEvents != 0) { + MaxEvents = other.MaxEvents; + } + _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(pb::CodedInputStream input) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + input.ReadRawMessage(this); + #else + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input); + break; + case 10: { + executionIds_.AddEntriesFrom(input, _repeated_executionIds_codec); + break; + } + case 18: { + if (sinceUtc_ == null) { + SinceUtc = new global::Google.Protobuf.WellKnownTypes.Timestamp(); + } + input.ReadMessage(SinceUtc); + break; + } + case 24: { + MaxEvents = input.ReadInt32(); + break; + } + } + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) { + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input); + break; + case 10: { + executionIds_.AddEntriesFrom(ref input, _repeated_executionIds_codec); + break; + } + case 18: { + if (sinceUtc_ == null) { + SinceUtc = new global::Google.Protobuf.WellKnownTypes.Timestamp(); + } + input.ReadMessage(SinceUtc); + break; + } + case 24: { + MaxEvents = input.ReadInt32(); + break; + } + } + } + } + #endif + + } + + [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] + public sealed partial class PullTraceEventsResponse : pb::IMessage + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + , pb::IBufferMessage + #endif + { + private static readonly pb::MessageParser _parser = new pb::MessageParser(() => new PullTraceEventsResponse()); + private pb::UnknownFieldSet _unknownFields; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pb::MessageParser Parser { get { return _parser; } } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pbr::MessageDescriptor Descriptor { + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[15]; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + pbr::MessageDescriptor pb::IMessage.Descriptor { + get { return Descriptor; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullTraceEventsResponse() { + OnConstruction(); + } + + partial void OnConstruction(); + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullTraceEventsResponse(PullTraceEventsResponse other) : this() { + events_ = other.events_.Clone(); + _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public PullTraceEventsResponse Clone() { + return new PullTraceEventsResponse(this); + } + + /// Field number for the "events" field. + public const int EventsFieldNumber = 1; + private static readonly pb::FieldCodec _repeated_events_codec + = pb::FieldCodec.ForMessage(10, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventDto.Parser); + private readonly pbc::RepeatedField events_ = new pbc::RepeatedField(); + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public pbc::RepeatedField Events { + get { return events_; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override bool Equals(object other) { + return Equals(other as PullTraceEventsResponse); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public bool Equals(PullTraceEventsResponse other) { + if (ReferenceEquals(other, null)) { + return false; + } + if (ReferenceEquals(other, this)) { + return true; + } + if(!events_.Equals(other.events_)) return false; + return Equals(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override int GetHashCode() { + int hash = 1; + hash ^= events_.GetHashCode(); + if (_unknownFields != null) { + hash ^= _unknownFields.GetHashCode(); + } + return hash; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override string ToString() { + return pb::JsonFormatter.ToDiagnosticString(this); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void WriteTo(pb::CodedOutputStream output) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + output.WriteRawMessage(this); + #else + events_.WriteTo(output, _repeated_events_codec); + if (_unknownFields != null) { + _unknownFields.WriteTo(output); + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalWriteTo(ref pb::WriteContext output) { + events_.WriteTo(ref output, _repeated_events_codec); + if (_unknownFields != null) { + _unknownFields.WriteTo(ref output); + } + } + #endif + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public int CalculateSize() { + int size = 0; + size += events_.CalculateSize(_repeated_events_codec); + if (_unknownFields != null) { + size += _unknownFields.CalculateSize(); + } + return size; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(PullTraceEventsResponse other) { + if (other == null) { + return; + } + events_.Add(other.events_); + _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(pb::CodedInputStream input) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + input.ReadRawMessage(this); + #else + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input); + break; + case 10: { + events_.AddEntriesFrom(input, _repeated_events_codec); + break; + } + } + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) { + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input); + break; + case 10: { + events_.AddEntriesFrom(ref input, _repeated_events_codec); + break; + } + } + } + } + #endif + + } + /// /// Site Call Audit (#22) reconciliation pull: central→site request for any /// site-local operation-tracking rows whose UpdatedAtUtc >= since_utc — the @@ -5837,7 +6391,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[14]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[16]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -6127,7 +6681,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[15]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[17]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/SitestreamGrpc.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/SitestreamGrpc.cs index 6db5da2a..14789b37 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/SitestreamGrpc.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/SitestreamGrpc.cs @@ -62,6 +62,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Marshaller __Marshaller_sitestream_PullAuditEventsResponse = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse.Parser)); [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + static readonly grpc::Marshaller __Marshaller_sitestream_PullTraceEventsRequest = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest.Parser)); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + static readonly grpc::Marshaller __Marshaller_sitestream_PullTraceEventsResponse = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsResponse.Parser)); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Marshaller __Marshaller_sitestream_PullSiteCallsRequest = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest.Parser)); [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Marshaller __Marshaller_sitestream_PullSiteCallsResponse = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse.Parser)); @@ -106,6 +110,14 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { __Marshaller_sitestream_PullAuditEventsRequest, __Marshaller_sitestream_PullAuditEventsResponse); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + static readonly grpc::Method __Method_PullTraceEvents = new grpc::Method( + grpc::MethodType.Unary, + __ServiceName, + "PullTraceEvents", + __Marshaller_sitestream_PullTraceEventsRequest, + __Marshaller_sitestream_PullTraceEventsResponse); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Method __Method_PullSiteCalls = new grpc::Method( grpc::MethodType.Unary, @@ -163,6 +175,20 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { throw new grpc::RpcException(new grpc::Status(grpc::StatusCode.Unimplemented, "")); } + /// + /// Flow Debugger (design 2026-09-05 §5.4): point read of a site's ScriptTrace rows for a + /// set of execution ids (by ExecutionId OR ParentExecutionId). NOT a reconciliation path — + /// moves no cursor, flips no ForwardState. Same PSK interceptor as PullAuditEvents. + /// + /// The request received from the client. + /// The context of the server-side call handler being invoked. + /// The response to send back to the client (wrapped by a task). + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual global::System.Threading.Tasks.Task PullTraceEvents(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest request, grpc::ServerCallContext context) + { + throw new grpc::RpcException(new grpc::Status(grpc::StatusCode.Unimplemented, "")); + } + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] public virtual global::System.Threading.Tasks.Task PullSiteCalls(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest request, grpc::ServerCallContext context) { @@ -296,6 +322,62 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { { return CallInvoker.AsyncUnaryCall(__Method_PullAuditEvents, null, options, request); } + /// + /// Flow Debugger (design 2026-09-05 §5.4): point read of a site's ScriptTrace rows for a + /// set of execution ids (by ExecutionId OR ParentExecutionId). NOT a reconciliation path — + /// moves no cursor, flips no ForwardState. Same PSK interceptor as PullAuditEvents. + /// + /// The request to send to the server. + /// The initial metadata to send with the call. This parameter is optional. + /// An optional deadline for the call. The call will be cancelled if deadline is hit. + /// An optional token for canceling the call. + /// The response received from the server. + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsResponse PullTraceEvents(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest request, grpc::Metadata headers = null, global::System.DateTime? deadline = null, global::System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) + { + return PullTraceEvents(request, new grpc::CallOptions(headers, deadline, cancellationToken)); + } + /// + /// Flow Debugger (design 2026-09-05 §5.4): point read of a site's ScriptTrace rows for a + /// set of execution ids (by ExecutionId OR ParentExecutionId). NOT a reconciliation path — + /// moves no cursor, flips no ForwardState. Same PSK interceptor as PullAuditEvents. + /// + /// The request to send to the server. + /// The options for the call. + /// The response received from the server. + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsResponse PullTraceEvents(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest request, grpc::CallOptions options) + { + return CallInvoker.BlockingUnaryCall(__Method_PullTraceEvents, null, options, request); + } + /// + /// Flow Debugger (design 2026-09-05 §5.4): point read of a site's ScriptTrace rows for a + /// set of execution ids (by ExecutionId OR ParentExecutionId). NOT a reconciliation path — + /// moves no cursor, flips no ForwardState. Same PSK interceptor as PullAuditEvents. + /// + /// The request to send to the server. + /// The initial metadata to send with the call. This parameter is optional. + /// An optional deadline for the call. The call will be cancelled if deadline is hit. + /// An optional token for canceling the call. + /// The call object. + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual grpc::AsyncUnaryCall PullTraceEventsAsync(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest request, grpc::Metadata headers = null, global::System.DateTime? deadline = null, global::System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) + { + return PullTraceEventsAsync(request, new grpc::CallOptions(headers, deadline, cancellationToken)); + } + /// + /// Flow Debugger (design 2026-09-05 §5.4): point read of a site's ScriptTrace rows for a + /// set of execution ids (by ExecutionId OR ParentExecutionId). NOT a reconciliation path — + /// moves no cursor, flips no ForwardState. Same PSK interceptor as PullAuditEvents. + /// + /// The request to send to the server. + /// The options for the call. + /// The call object. + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual grpc::AsyncUnaryCall PullTraceEventsAsync(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullTraceEventsRequest request, grpc::CallOptions options) + { + return CallInvoker.AsyncUnaryCall(__Method_PullTraceEvents, null, options, request); + } [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] public virtual global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse PullSiteCalls(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest request, grpc::Metadata headers = null, global::System.DateTime? deadline = null, global::System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) { @@ -335,6 +417,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { .AddMethod(__Method_IngestAuditEvents, serviceImpl.IngestAuditEvents) .AddMethod(__Method_IngestCachedTelemetry, serviceImpl.IngestCachedTelemetry) .AddMethod(__Method_PullAuditEvents, serviceImpl.PullAuditEvents) + .AddMethod(__Method_PullTraceEvents, serviceImpl.PullTraceEvents) .AddMethod(__Method_PullSiteCalls, serviceImpl.PullSiteCalls).Build(); } @@ -350,6 +433,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { serviceBinder.AddMethod(__Method_IngestAuditEvents, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.IngestAuditEvents)); serviceBinder.AddMethod(__Method_IngestCachedTelemetry, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.IngestCachedTelemetry)); serviceBinder.AddMethod(__Method_PullAuditEvents, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.PullAuditEvents)); + serviceBinder.AddMethod(__Method_PullTraceEvents, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.PullTraceEvents)); serviceBinder.AddMethod(__Method_PullSiteCalls, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.PullSiteCalls)); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditRetentionServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditRetentionServiceTests.cs index 57f333e4..52c5676a 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditRetentionServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditRetentionServiceTests.cs @@ -127,6 +127,10 @@ public class SiteAuditRetentionServiceTests /// Stub queue that records purge cutoffs and can throw on the first tick. private sealed class RecordingSiteAuditQueue : ISiteAuditQueue { + public Task> ReadByExecutionIdsAsync( + IReadOnlyCollection executionIds, DateTime sinceUtc, int limit, CancellationToken ct = default) + => throw new NotSupportedException("not exercised by the retention service"); + private readonly ConcurrentQueue _purgeCalls = new(); private int _calls; diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterReadByExecutionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterReadByExecutionTests.cs new file mode 100644 index 00000000..3deb2b4f --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterReadByExecutionTests.cs @@ -0,0 +1,83 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.AuditLog.Site; +using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site; + +/// +/// Flow Debugger (design 2026-09-05 §5.4): +/// matches on ExecutionId OR ParentExecutionId, honours the since bound, orders by +/// (OccurredAtUtc, EventId), caps the row count, and is independent of forwarding state. +/// +public class SqliteAuditWriterReadByExecutionTests : IDisposable +{ + private readonly string _dbPath = Path.Combine(Path.GetTempPath(), $"audit-readbyexec-{Guid.NewGuid():N}.db"); + private static readonly DateTime T0 = new(2026, 9, 5, 12, 0, 0, DateTimeKind.Utc); + + public void Dispose() + { + if (File.Exists(_dbPath)) { try { File.Delete(_dbPath); } catch { /* best-effort */ } } + } + + private SqliteAuditWriter CreateWriter() => new( + Options.Create(new SqliteAuditWriterOptions { DatabasePath = _dbPath }), + NullLogger.Instance, + new FakeNodeIdentityProvider()); + + private static AuditEvent Row(DateTime at, Guid? exec, Guid? parent, AuditKind kind = AuditKind.ScriptRun) => + ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, kind, AuditStatus.Delivered, + occurredAtUtc: at, executionId: exec, parentExecutionId: parent, sourceSiteId: "site-a"); + + [Fact] + public async Task MatchesExecutionId_OrParentExecutionId_OrderedAndBounded() + { + var root = Guid.NewGuid(); + var child = Guid.NewGuid(); + var unrelated = Guid.NewGuid(); + var rows = new[] + { + Row(T0.AddSeconds(3), child, root, AuditKind.AttributeWrite), // parent match + Row(T0.AddSeconds(1), root, null), // exec match + Row(T0.AddSeconds(2), child, root), // parent match + Row(T0.AddSeconds(4), unrelated, null), // no match + Row(T0.AddSeconds(-60), root, null), // before since + }; + + await using (var writer = CreateWriter()) + { + foreach (var r in rows) await writer.WriteAsync(r); + + var hits = await writer.ReadByExecutionIdsAsync([root], T0, limit: 100); + Assert.Equal(3, hits.Count); + Assert.Equal(rows[1].EventId, hits[0].EventId); + Assert.Equal(rows[2].EventId, hits[1].EventId); + Assert.Equal(rows[0].EventId, hits[2].EventId); + Assert.Equal(rows[1], hits[0]); + + // Bounded. + var two = await writer.ReadByExecutionIdsAsync([root], T0, limit: 2); + Assert.Equal(2, two.Count); + + // Forwarding state is irrelevant: mark everything Reconciled, still served. + await writer.MarkReconciledAsync(rows.Select(r => r.EventId).ToList()); + Assert.Equal(3, (await writer.ReadByExecutionIdsAsync([root], T0, limit: 100)).Count); + + // Empty id set → empty; multiple ids union. + Assert.Empty(await writer.ReadByExecutionIdsAsync(Array.Empty(), T0, 10)); + Assert.Equal(4, (await writer.ReadByExecutionIdsAsync([root, unrelated], T0, 100)).Count); + } + } + + [Fact] + public async Task Guards_LimitAndIdCap() + { + await using var writer = CreateWriter(); + await Assert.ThrowsAsync(() => writer.ReadByExecutionIdsAsync([Guid.NewGuid()], T0, 0)); + var tooMany = Enumerable.Range(0, SqliteAuditWriter.MaxExecutionIdsPerRead + 1).Select(_ => Guid.NewGuid()).ToList(); + await Assert.ThrowsAsync(() => writer.ReadByExecutionIdsAsync(tooMany, T0, 10)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullTraceEventsTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullTraceEventsTests.cs new file mode 100644 index 00000000..230a37e9 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullTraceEventsTests.cs @@ -0,0 +1,125 @@ +using Akka.TestKit.Xunit2; +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +/// +/// Flow Debugger (design 2026-09-05 §5.4): +/// — request validation, the queue point-read, DTO projection, and the guarantee that it is +/// NOT a reconciliation path (no cursor flip). Also pins that the RPC sits under the +/// PSK-gated /sitestream.SiteStreamService/ prefix. +/// +public class SiteStreamPullTraceEventsTests : TestKit +{ + private readonly ISiteStreamSubscriber _subscriber = Substitute.For(); + + private SiteStreamGrpcServer CreateServer() => new(_subscriber, NullLogger.Instance); + + private static ServerCallContext NewContext() + { + var context = Substitute.For(); + context.CancellationToken.Returns(CancellationToken.None); + return context; + } + + private static AuditEvent Row(Guid executionId) => ScadaBridgeAuditEventFactory.Create( + AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, + occurredAtUtc: new DateTime(2026, 9, 5, 12, 0, 0, DateTimeKind.Utc), executionId: executionId, sourceSiteId: "site-a"); + + [Fact] + public void PullTraceEvents_IsUnderThePskGatedServicePrefix() + { + var method = SiteStreamService.Descriptor.FindMethodByName("PullTraceEvents"); + Assert.NotNull(method); + Assert.Equal("sitestream.SiteStreamService", SiteStreamService.Descriptor.FullName); + } + + [Fact] + public async Task NoQueueWired_ReturnsEmpty() + { + var response = await CreateServer().PullTraceEvents(new PullTraceEventsRequest + { + ExecutionIds = { Guid.NewGuid().ToString("D") }, MaxEvents = 10, + }, NewContext()); + Assert.Empty(response.Events); + } + + [Fact] + public async Task ReadsByIds_ProjectsDtos_AndNeverFlipsState() + { + var queue = Substitute.For(); + var id = Guid.NewGuid(); + var rows = new List { Row(id), Row(id) }; + IReadOnlyCollection? seenIds = null; + DateTime seenSince = default; + int seenLimit = 0; + queue.ReadByExecutionIdsAsync( + Arg.Do>(x => seenIds = x), Arg.Do(x => seenSince = x), + Arg.Do(x => seenLimit = x), Arg.Any()) + .Returns((IReadOnlyList)rows); + var server = CreateServer(); + server.SetSiteAuditQueue(queue); + var since = new DateTime(2026, 9, 5, 11, 0, 0, DateTimeKind.Utc); + + var response = await server.PullTraceEvents(new PullTraceEventsRequest + { + ExecutionIds = { id.ToString("D") }, SinceUtc = Timestamp.FromDateTime(since), MaxEvents = 500, + }, NewContext()); + + Assert.Equal(2, response.Events.Count); + Assert.Equal(rows[0].EventId.ToString(), response.Events[0].EventId); + Assert.Equal(new[] { id }, seenIds); + Assert.Equal(since, seenSince); + Assert.Equal(500, seenLimit); + await queue.DidNotReceiveWithAnyArgs().MarkReconciledUpToAsync(default, default, default); + await queue.DidNotReceiveWithAnyArgs().MarkReconciledAsync(default!, default); + } + + [Theory] + [InlineData(0, 1)] + [InlineData(2001, 1)] + [InlineData(10, 0)] + [InlineData(10, 65)] + public async Task InvalidArguments_AreRejected(int maxEvents, int idCount) + { + var server = CreateServer(); + server.SetSiteAuditQueue(Substitute.For()); + var request = new PullTraceEventsRequest { MaxEvents = maxEvents }; + for (var i = 0; i < idCount; i++) request.ExecutionIds.Add(Guid.NewGuid().ToString("D")); + + var ex = await Assert.ThrowsAsync(() => server.PullTraceEvents(request, NewContext())); + Assert.Equal(StatusCode.InvalidArgument, ex.StatusCode); + } + + [Fact] + public async Task NonGuidId_IsRejected() + { + var server = CreateServer(); + server.SetSiteAuditQueue(Substitute.For()); + var ex = await Assert.ThrowsAsync(() => server.PullTraceEvents( + new PullTraceEventsRequest { ExecutionIds = { "nope" }, MaxEvents = 10 }, NewContext())); + Assert.Equal(StatusCode.InvalidArgument, ex.StatusCode); + } + + [Fact] + public async Task QueueFault_CollapsesToEmpty() + { + var queue = Substitute.For(); + queue.ReadByExecutionIdsAsync(default!, default, default, default) + .ReturnsForAnyArgs>(_ => throw new InvalidOperationException("boom")); + var server = CreateServer(); + server.SetSiteAuditQueue(queue); + + var response = await server.PullTraceEvents( + new PullTraceEventsRequest { ExecutionIds = { Guid.NewGuid().ToString("D") }, MaxEvents = 10 }, NewContext()); + Assert.Empty(response.Events); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/StreamRelayTraceStampTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/StreamRelayTraceStampTests.cs new file mode 100644 index 00000000..7f3c807c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/StreamRelayTraceStampTests.cs @@ -0,0 +1,76 @@ +using System.Threading.Channels; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Google.Protobuf.WellKnownTypes; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +/// +/// Flow Debugger (design 2026-09-05 §5.3/§5.4): AttributeValueUpdate fields 7/8 — +/// stamps source_execution_id/trace on the way +/// out and parses them back; an unstamped change (and an +/// older site's frame) round-trips to null/false. +/// +public class StreamRelayTraceStampTests : TestKit +{ + private static SiteStreamEvent Relay(AttributeValueChanged change, ActorSystem sys) + { + var channel = Channel.CreateUnbounded(); + var actor = sys.ActorOf(Props.Create(() => new StreamRelayActor("corr", channel.Writer))); + actor.Tell(change); + SiteStreamEvent? evt = null; + for (var i = 0; i < 40 && !channel.Reader.TryRead(out evt); i++) Thread.Sleep(50); + Assert.NotNull(evt); + return evt!; + } + + [Fact] + public void TracedChange_IsStampedOnTheWire_AndParsedBack() + { + var source = Guid.NewGuid(); + var now = new DateTimeOffset(2026, 9, 5, 12, 0, 0, TimeSpan.Zero); + var change = new AttributeValueChanged("P1", "Status", "Status", "Running", "Good", now, SourceExecutionId: source, Trace: true); + + var evt = Relay(change, Sys); + Assert.Equal(source.ToString("D"), evt.AttributeChanged.SourceExecutionId); + Assert.True(evt.AttributeChanged.Trace); + + var back = Assert.IsType(SiteStreamGrpcClient.ConvertToDomainEvent(evt)); + Assert.Equal(source, back.SourceExecutionId); + Assert.True(back.Trace); + } + + [Fact] + public void UnstampedChange_RoundTripsToNullAndFalse() + { + var change = new AttributeValueChanged("P1", "Status", "Status", "Idle", "Good", DateTimeOffset.UtcNow); + + var evt = Relay(change, Sys); + Assert.Equal(string.Empty, evt.AttributeChanged.SourceExecutionId); + Assert.False(evt.AttributeChanged.Trace); + + var back = Assert.IsType(SiteStreamGrpcClient.ConvertToDomainEvent(evt)); + Assert.Null(back.SourceExecutionId); + Assert.False(back.Trace); + } + + [Fact] + public void OlderSiteFrame_WithoutTheFields_ParsesToNullAndFalse() + { + var evt = new SiteStreamEvent + { + CorrelationId = "c", + AttributeChanged = new AttributeValueUpdate + { + InstanceUniqueName = "P1", AttributePath = "A", AttributeName = "A", Value = "1", + Quality = Quality.Good, Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow), + }, + }; + var back = Assert.IsType(SiteStreamGrpcClient.ConvertToDomainEvent(evt)); + Assert.Null(back.SourceExecutionId); + Assert.False(back.Trace); + } +}