feat(wire): attribute updates carry source_execution_id/trace; PullTraceEvents RPC + site read-by-execution-ids

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
2026-09-05 23:52:50 -04:00
parent f15d0db0ad
commit 7e61552d82
15 changed files with 1458 additions and 80 deletions
@@ -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;
+
+/// <summary>
+/// Production <see cref="IPullTraceEventsClient"/> over the site-hosted <c>PullTraceEvents</c>
+/// unary RPC (Flow Debugger design 2026-09-05 §5.4). Mirrors
+/// <see cref="GrpcPullAuditEventsClient"/> exactly: endpoint resolution through
+/// <see cref="ISiteEnumerator"/> 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.
+/// </summary>
+public sealed class GrpcPullTraceEventsClient : IPullTraceEventsClient
+{
+    /// <summary>Hard cap on ids per call — matches the site's <c>SqliteAuditWriter.MaxExecutionIdsPerRead</c>.</summary>
+    public const int MaxExecutionIds = 64;
+
+    private readonly ISiteEnumerator _sites;
+    private readonly IPullTraceEventsInvoker _invoker;
+    private readonly ILogger<GrpcPullTraceEventsClient> _logger;
+
+    /// <summary>Creates the client over the given site enumerator and unary-call invoker.</summary>
+    public GrpcPullTraceEventsClient(
+        ISiteEnumerator sites,
+        IPullTraceEventsInvoker invoker,
+        ILogger<GrpcPullTraceEventsClient> logger)
+    {
+        _sites = sites ?? throw new ArgumentNullException(nameof(sites));
+        _invoker = invoker ?? throw new ArgumentNullException(nameof(invoker));
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+    }
+
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<AuditEvent>> PullAsync(
+        string siteId,
+        IReadOnlyCollection<Guid> executionIds,
+        DateTime sinceUtc,
+        int maxEvents,
+        CancellationToken ct)
+    {
+        ArgumentNullException.ThrowIfNull(executionIds);
+        if (executionIds.Count == 0 || maxEvents <= 0)
+        {
+            return Array.Empty<AuditEvent>();
+        }
+        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<AuditEvent>();
+        }
+
+        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<AuditEvent>();
+        }
+
+        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);
+
+    /// <summary>
+    /// Seam over the <c>PullTraceEvents</c> unary gRPC call. Production binds
+    /// <see cref="GrpcPullTraceEventsInvoker"/>; tests inject a fake.
+    /// </summary>
+    public interface IPullTraceEventsInvoker
+    {
+        /// <summary>Issues the RPC against <paramref name="endpoint"/> with the site's preshared key.</summary>
+        Task<PullTraceEventsResponse> InvokeAsync(
+            string siteId, string endpoint, PullTraceEventsRequest request, CancellationToken ct);
+    }
+}
+
+/// <summary>
+/// Production <see cref="GrpcPullTraceEventsClient.IPullTraceEventsInvoker"/>: one cached
+/// <see cref="GrpcChannel"/> per (site, endpoint), keepalive from <see cref="CommunicationOptions"/>,
+/// per-site PSK call credentials — the same shape as <see cref="GrpcPullAuditEventsInvoker"/>.
+/// </summary>
+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;
+
+    /// <summary>Creates the invoker with the given options and optional per-site PSK provider.</summary>
+    public GrpcPullTraceEventsInvoker(CommunicationOptions options, ISitePskProvider? pskProvider)
+    {
+        _options = options ?? throw new ArgumentNullException(nameof(options));
+        _pskProvider = pskProvider;
+    }
+
+    /// <inheritdoc />
+    public async Task<PullTraceEventsResponse> 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));
+
+    /// <inheritdoc />
+    public void Dispose()
+    {
+        foreach (var channel in _channels.Values)
+        {
+            channel.Dispose();
+        }
+        _channels.Clear();
+    }
+}
@@ -0,0 +1,27 @@
+using ZB.MOM.WW.Audit;
+
+namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
+
+/// <summary>
+/// Flow Debugger (design 2026-09-05 §5.4): central-side client for the site-hosted
+/// <c>PullTraceEvents</c> unary RPC — a point read of a site's <c>ScriptTrace</c> rows for a
+/// set of execution ids (by <c>ExecutionId</c> or <c>ParentExecutionId</c>). 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.
+/// </summary>
+public interface IPullTraceEventsClient
+{
+    /// <summary>Pulls the site's trace rows for <paramref name="executionIds"/> (max 64) since <paramref name="sinceUtc"/>.</summary>
+    /// <param name="siteId">The site to pull from (resolved to its gRPC endpoint + preshared key).</param>
+    /// <param name="executionIds">Execution ids to match on <c>ExecutionId</c> or <c>ParentExecutionId</c>.</param>
+    /// <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>
+    /// <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);
+}
@@ -551,6 +551,26 @@ public static class ServiceCollectionExtensions
            sp.GetRequiredService<GrpcPullAuditEventsClient.IPullAuditEventsInvoker>(),
            sp.GetRequiredService<ILogger<GrpcPullAuditEventsClient>>()));

+        // 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<GrpcPullTraceEventsInvoker>(sp =>
+        {
+            var options = sp
+                .GetService<Microsoft.Extensions.Options.IOptions<
+                    ZB.MOM.WW.ScadaBridge.Communication.CommunicationOptions>>();
+            var psk = sp.GetService<ZB.MOM.WW.ScadaBridge.Communication.Grpc.ISitePskProvider>();
+            return options is null
+                ? new GrpcPullTraceEventsInvoker(new ZB.MOM.WW.ScadaBridge.Communication.CommunicationOptions(), psk)
+                : new GrpcPullTraceEventsInvoker(options.Value, psk);
+        });
+        services.TryAddSingleton<GrpcPullTraceEventsClient.IPullTraceEventsInvoker>(
+            sp => sp.GetRequiredService<GrpcPullTraceEventsInvoker>());
+        services.TryAddSingleton<IPullTraceEventsClient>(sp => new GrpcPullTraceEventsClient(
+            sp.GetRequiredService<ISiteEnumerator>(),
+            sp.GetRequiredService<GrpcPullTraceEventsClient.IPullTraceEventsInvoker>(),
+            sp.GetRequiredService<ILogger<GrpcPullTraceEventsClient>>()));
+
        // 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
@@ -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
        }
    }

+    /// <summary>Flow Debugger PullTraceEvents: hard cap on ids per read (design §5.4).</summary>
+    public const int MaxExecutionIdsPerRead = 64;
+
+    /// <inheritdoc />
+    public Task<IReadOnlyList<AuditEvent>> ReadByExecutionIdsAsync(
+        IReadOnlyCollection<Guid> 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<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>());
+        }
+        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<string>(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<AuditEvent>(Math.Min(limit, 256));
+            using var reader = cmd.ExecuteReader();
+            while (reader.Read())
+            {
+                rows.Add(MapRow(reader));
+            }
+            return Task.FromResult<IReadOnlyList<AuditEvent>>(rows);
+        }
+    }
+
    /// <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