feat(mesh): node gRPC artifact fetcher — SHA-verified reassembly, endpoint failover, null-on-any-failure

Phase 3 Task 4. GrpcDeploymentArtifactFetcher streams the artifact from the first configured central endpoint that answers, reassembles the chunks, and verifies SHA-256(bytes) lowercase-hex == the dispatched RevisionHash (byte-identical to ConfigComposer's Convert.ToHexStringLower(SHA256.HashData(blob))). Every per-endpoint problem — RpcException, zero-length stream, or hash mismatch — is logged and the next endpoint tried; if none yield verified bytes it returns null (apply failure, keep last-known-good — the #485 contract), never throwing into the actor loop. A Func<endpoint, client> seam keeps the gRPC boundary out of the unit tests (that is Task 8's job); the real path caches one h2c GrpcChannel per endpoint. Files live under the .DeploymentCache namespace (folder Deployment/) to avoid shadowing the Configuration.Entities.Deployment type in the test assembly — same convention as LocalDbDeploymentArtifactCache. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-22 19:28:48 -04:00
parent 4d88f67c0d
commit 7a8fb5f129
4 changed files with 346 additions and 0 deletions
@@ -0,0 +1,153 @@
+using System.Collections.Concurrent;
+using System.Security.Cryptography;
+using Grpc.Core;
+using Grpc.Net.Client;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.OtOpcUa.Cluster;
+using ZB.MOM.WW.OtOpcUa.Commons.Protos.DeploymentArtifact.V1;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
+
+/// <summary>
+///     Streams the deployed-configuration artifact from central over gRPC and verifies it against the
+///     revision hash the dispatch already carries (per-cluster mesh Phase 3, <c>FetchAndCache</c>).
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>One invariant.</b> Return the first endpoint's bytes that both reassemble AND verify
+///         (<c>SHA-256(bytes)</c> lowercase-hex == the expected revision hash); otherwise
+///         <see langword="null"/>. Every per-endpoint problem — an <see cref="RpcException"/>
+///         (Unavailable / NotFound / deadline), a zero-length stream, or a hash mismatch — is treated
+///         uniformly as "this endpoint did not supply valid bytes," logged, and the next endpoint is
+///         tried. A hash mismatch is tried-next rather than fatal because a partitioned central could
+///         legitimately serve stale bytes from one node and current bytes from another; when every
+///         central shares one SQL store (the common case) the retry is merely cheap and harmless.
+///     </para>
+///     <para>
+///         <b>Why <see langword="null"/>, not throw.</b> The caller treats <see langword="null"/>
+///         identically to today's "reconcile returned null": an apply failure that keeps
+///         last-known-good and does not advance the revision. A typed "no bytes" keeps the #485
+///         contract explicit and cannot escape as an unhandled actor message.
+///     </para>
+/// </remarks>
+public sealed class GrpcDeploymentArtifactFetcher : IDeploymentArtifactFetcher, IDisposable
+{
+    private readonly ConfigSourceOptions _options;
+    private readonly ILogger<GrpcDeploymentArtifactFetcher> _log;
+    private readonly Func<string, DeploymentArtifactService.DeploymentArtifactServiceClient> _clientFactory;
+    private readonly ConcurrentDictionary<string, GrpcChannel> _channels = new(StringComparer.Ordinal);
+
+    /// <summary>Initializes a new instance of the <see cref="GrpcDeploymentArtifactFetcher"/> class.</summary>
+    /// <param name="options">Config-source options; supplies the endpoints, shared key, and deadline.</param>
+    /// <param name="log">Logger.</param>
+    /// <param name="clientFactory">
+    ///     Seam for tests: builds the generated client for an endpoint. Defaults to a real h2c
+    ///     <see cref="GrpcChannel"/> (one cached per endpoint).
+    /// </param>
+    public GrpcDeploymentArtifactFetcher(
+        IOptions<ConfigSourceOptions> options,
+        ILogger<GrpcDeploymentArtifactFetcher> log,
+        Func<string, DeploymentArtifactService.DeploymentArtifactServiceClient>? clientFactory = null)
+    {
+        ArgumentNullException.ThrowIfNull(options);
+        ArgumentNullException.ThrowIfNull(log);
+
+        _options = options.Value;
+        _log = log;
+        _clientFactory = clientFactory ?? DefaultClientFactory;
+    }
+
+    /// <inheritdoc />
+    public async Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(deploymentId);
+        ArgumentException.ThrowIfNullOrEmpty(expectedRevisionHash);
+
+        var endpoints = _options.CentralFetchEndpoints;
+        if (endpoints.Length == 0)
+        {
+            _log.LogWarning(
+                "Artifact fetch for {DeploymentId} skipped: no ConfigSource:CentralFetchEndpoints configured.",
+                deploymentId);
+            return null;
+        }
+
+        foreach (var endpoint in endpoints)
+        {
+            try
+            {
+                var bytes = await FetchFromEndpointAsync(endpoint, deploymentId, ct);
+                if (bytes.Length == 0)
+                {
+                    _log.LogWarning(
+                        "Artifact fetch for {DeploymentId} from {Endpoint} returned zero bytes; trying next endpoint.",
+                        deploymentId, endpoint);
+                    continue;
+                }
+
+                var actualHash = Convert.ToHexStringLower(SHA256.HashData(bytes));
+                if (!string.Equals(actualHash, expectedRevisionHash, StringComparison.OrdinalIgnoreCase))
+                {
+                    _log.LogWarning(
+                        "Artifact fetch for {DeploymentId} from {Endpoint} failed SHA-256 verification " +
+                        "(expected {Expected}, got {Actual}); trying next endpoint.",
+                        deploymentId, endpoint, expectedRevisionHash, actualHash);
+                    continue;
+                }
+
+                _log.LogInformation(
+                    "Fetched + verified artifact for {DeploymentId} from {Endpoint} ({Bytes} bytes).",
+                    deploymentId, endpoint, bytes.Length);
+                return bytes;
+            }
+            catch (RpcException ex)
+            {
+                _log.LogWarning(
+                    "Artifact fetch for {DeploymentId} from {Endpoint} failed ({Status}); trying next endpoint.",
+                    deploymentId, endpoint, ex.StatusCode);
+            }
+        }
+
+        _log.LogWarning(
+            "Artifact fetch for {DeploymentId} failed on ALL {Count} endpoint(s); returning no bytes " +
+            "(apply failure — last-known-good is kept).",
+            deploymentId, endpoints.Length);
+        return null;
+    }
+
+    private async Task<byte[]> FetchFromEndpointAsync(string endpoint, string deploymentId, CancellationToken ct)
+    {
+        var client = _clientFactory(endpoint);
+        var headers = new Metadata { { "authorization", $"Bearer {_options.ApiKey}" } };
+
+        using var deadlineCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        if (_options.FetchTimeoutSeconds > 0)
+            deadlineCts.CancelAfter(TimeSpan.FromSeconds(_options.FetchTimeoutSeconds));
+
+        using var call = client.Fetch(
+            new FetchRequest { DeploymentId = deploymentId }, headers, cancellationToken: deadlineCts.Token);
+
+        using var buffer = new MemoryStream();
+        await foreach (var chunk in call.ResponseStream.ReadAllAsync(deadlineCts.Token))
+            chunk.Data.WriteTo(buffer);
+
+        return buffer.ToArray();
+    }
+
+    private DeploymentArtifactService.DeploymentArtifactServiceClient DefaultClientFactory(string endpoint)
+    {
+        // h2c: GrpcChannel over an http:// address negotiates prior-knowledge HTTP/2. One channel per
+        // endpoint, cached and reused (creating a channel per call would leak sockets).
+        var channel = _channels.GetOrAdd(endpoint, ep => GrpcChannel.ForAddress(ep));
+        return new DeploymentArtifactService.DeploymentArtifactServiceClient(channel);
+    }
+
+    /// <inheritdoc />
+    public void Dispose()
+    {
+        foreach (var channel in _channels.Values)
+            channel.Dispose();
+        _channels.Clear();
+    }
+}
@@ -0,0 +1,28 @@
+namespace ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
+
+/// <summary>
+///     Fetches the deployed-configuration artifact bytes from central over gRPC (per-cluster mesh
+///     Phase 3, <c>FetchAndCache</c> mode). The node-side counterpart of central's
+///     <c>DeploymentArtifactService</c>.
+/// </summary>
+public interface IDeploymentArtifactFetcher
+{
+    /// <summary>
+    ///     Fetches and reassembles the artifact for <paramref name="deploymentId"/> from the first
+    ///     configured central endpoint that answers, verifying that
+    ///     <c>SHA-256(reassembled bytes)</c> (lowercase hex) equals
+    ///     <paramref name="expectedRevisionHash"/> — the hash already carried in the dispatch.
+    /// </summary>
+    /// <param name="deploymentId">The deployment id to fetch.</param>
+    /// <param name="expectedRevisionHash">
+    ///     The lowercase-hex SHA-256 the reassembled bytes must match (<c>Deployment.RevisionHash</c>).
+    /// </param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>
+    ///     The verified artifact bytes, or <see langword="null"/> on ANY failure — every configured
+    ///     endpoint unreachable/NotFound, a zero-length stream, or a SHA-256 mismatch. A
+    ///     <see langword="null"/> is "no answer," NEVER an empty configuration: the caller treats it as
+    ///     an apply failure and keeps last-known-good (the #485 contract).
+    /// </returns>
+    Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct);
+}
@@ -8,6 +8,9 @@
    <ItemGroup>
        <PackageReference Include="Akka.Hosting"/>
        <PackageReference Include="Akka.Cluster.Tools"/>
+        <!-- Node-side gRPC client that fetches the deployed-configuration artifact from central
+             (per-cluster mesh Phase 3, FetchAndCache mode). The generated stub lives in Commons. -->
+        <PackageReference Include="Grpc.Net.Client"/>
        <!-- Core LocalDb only (ILocalDb + the connection/transaction seam) — the deployment-artifact
             cache lives here, but the sync engine and its gRPC endpoint are the Host's concern. -->
        <PackageReference Include="ZB.MOM.WW.LocalDb" />
@@ -0,0 +1,162 @@
+using System.Security.Cryptography;
+using Google.Protobuf;
+using Grpc.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Cluster;
+using ZB.MOM.WW.OtOpcUa.Commons.Protos.DeploymentArtifact.V1;
+using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.DeploymentCache;
+
+/// <summary>
+///     Per-cluster mesh Phase 3 (Task 4) — the node-side gRPC artifact fetcher's logic: reassembly,
+///     SHA-256 verification, endpoint failover, and the null-on-any-failure (#485) contract. The real
+///     gRPC stream across a real Kestrel h2c listener is the Task 8 boundary test; here the client is
+///     a fake streaming canned chunks, so only the fetcher's own logic is under test.
+/// </summary>
+public sealed class GrpcDeploymentArtifactFetcherTests
+{
+    private const string DeploymentId = "11111111-1111-1111-1111-111111111111";
+
+    private static string Sha256Hex(byte[] bytes) => Convert.ToHexStringLower(SHA256.HashData(bytes));
+
+    private static GrpcDeploymentArtifactFetcher Fetcher(
+        string[] endpoints, Func<string, DeploymentArtifactService.DeploymentArtifactServiceClient> factory)
+        => new(
+            Options.Create(new ConfigSourceOptions
+            {
+                Mode = ConfigSourceOptions.ModeFetchAndCache,
+                CentralFetchEndpoints = endpoints,
+                ApiKey = "k",
+                FetchTimeoutSeconds = 30,
+            }),
+            NullLogger<GrpcDeploymentArtifactFetcher>.Instance,
+            factory);
+
+    [Fact]
+    public async Task Chunks_reassembling_to_the_expected_hash_return_the_exact_bytes()
+    {
+        var blob = new byte[300 * 1024];
+        for (var i = 0; i < blob.Length; i++) blob[i] = (byte)(i % 251);
+        var factory = SingleClient(StreamOf(blob, chunkSize: 128 * 1024));
+
+        var result = await Fetcher(["http://central-1:4055"], factory)
+            .FetchAsync(DeploymentId, Sha256Hex(blob), CancellationToken.None);
+
+        result.ShouldBe(blob);
+    }
+
+    [Fact]
+    public async Task A_hash_mismatch_returns_null()
+    {
+        var blob = new byte[] { 1, 2, 3, 4 };
+        var factory = SingleClient(StreamOf(blob, chunkSize: 2));
+
+        var result = await Fetcher(["http://central-1:4055"], factory)
+            .FetchAsync(DeploymentId, Sha256Hex([9, 9, 9, 9]), CancellationToken.None);
+
+        result.ShouldBeNull();
+    }
+
+    [Fact]
+    public async Task An_empty_stream_returns_null()
+    {
+        var factory = SingleClient(StreamOf([], chunkSize: 128 * 1024));
+
+        var result = await Fetcher(["http://central-1:4055"], factory)
+            .FetchAsync(DeploymentId, Sha256Hex([]), CancellationToken.None);
+
+        result.ShouldBeNull();
+    }
+
+    [Fact]
+    public async Task When_the_first_endpoint_is_unavailable_it_fails_over_to_the_second()
+    {
+        var blob = new byte[] { 10, 20, 30, 40, 50 };
+        var consulted = new List<string>();
+        var factory = (string endpoint) =>
+        {
+            consulted.Add(endpoint);
+            return endpoint.Contains("central-1")
+                ? (DeploymentArtifactService.DeploymentArtifactServiceClient)
+                    new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down")))
+                : new FakeClient(StreamOf(blob, chunkSize: 2));
+        };
+
+        var result = await Fetcher(["http://central-1:4055", "http://central-2:4055"], factory)
+            .FetchAsync(DeploymentId, Sha256Hex(blob), CancellationToken.None);
+
+        result.ShouldBe(blob);
+        // The failover must have actually consulted the second endpoint.
+        consulted.ShouldBe(["http://central-1:4055", "http://central-2:4055"]);
+    }
+
+    [Fact]
+    public async Task When_every_endpoint_throws_it_returns_null()
+    {
+        var factory = (string _) =>
+            (DeploymentArtifactService.DeploymentArtifactServiceClient)
+                new ThrowingClient(new RpcException(new Status(StatusCode.NotFound, "unknown deployment")));
+
+        var result = await Fetcher(["http://central-1:4055", "http://central-2:4055"], factory)
+            .FetchAsync(DeploymentId, Sha256Hex([1]), CancellationToken.None);
+
+        result.ShouldBeNull();
+    }
+
+    // ---- fakes --------------------------------------------------------------------------------
+
+    private static Func<string, DeploymentArtifactService.DeploymentArtifactServiceClient> SingleClient(
+        IReadOnlyList<ArtifactChunk> chunks)
+        => _ => new FakeClient(chunks);
+
+    private static IReadOnlyList<ArtifactChunk> StreamOf(byte[] blob, int chunkSize)
+    {
+        var chunks = new List<ArtifactChunk>();
+        for (var offset = 0; offset < blob.Length; offset += chunkSize)
+        {
+            var len = Math.Min(chunkSize, blob.Length - offset);
+            chunks.Add(new ArtifactChunk { Data = ByteString.CopyFrom(blob, offset, len) });
+        }
+
+        return chunks;
+    }
+
+    /// <summary>A generated client double whose <c>Fetch</c> streams canned chunks.</summary>
+    private sealed class FakeClient(IReadOnlyList<ArtifactChunk> chunks)
+        : DeploymentArtifactService.DeploymentArtifactServiceClient
+    {
+        public override AsyncServerStreamingCall<ArtifactChunk> Fetch(FetchRequest request, CallOptions options)
+            => new(
+                new ListStreamReader(chunks),
+                Task.FromResult(new Metadata()),
+                () => Status.DefaultSuccess,
+                () => [],
+                () => { });
+    }
+
+    /// <summary>A generated client double whose <c>Fetch</c> throws (endpoint down / NotFound).</summary>
+    private sealed class ThrowingClient(RpcException toThrow)
+        : DeploymentArtifactService.DeploymentArtifactServiceClient
+    {
+        public override AsyncServerStreamingCall<ArtifactChunk> Fetch(FetchRequest request, CallOptions options)
+            => throw toThrow;
+    }
+
+    private sealed class ListStreamReader(IReadOnlyList<ArtifactChunk> chunks) : IAsyncStreamReader<ArtifactChunk>
+    {
+        private int _index = -1;
+
+        public ArtifactChunk Current => chunks[_index];
+
+        public Task<bool> MoveNext(CancellationToken cancellationToken)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            _index++;
+            return Task.FromResult(_index < chunks.Count);
+        }
+    }
+}