diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/GrpcDeploymentArtifactFetcher.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/GrpcDeploymentArtifactFetcher.cs new file mode 100644 index 00000000..54b47cf0 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/GrpcDeploymentArtifactFetcher.cs @@ -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; + +/// +/// 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, FetchAndCache). +/// +/// +/// +/// One invariant. Return the first endpoint's bytes that both reassemble AND verify +/// (SHA-256(bytes) lowercase-hex == the expected revision hash); otherwise +/// . Every per-endpoint problem — an +/// (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. +/// +/// +/// Why , not throw. The caller treats +/// 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. +/// +/// +public sealed class GrpcDeploymentArtifactFetcher : IDeploymentArtifactFetcher, IDisposable +{ + private readonly ConfigSourceOptions _options; + private readonly ILogger _log; + private readonly Func _clientFactory; + private readonly ConcurrentDictionary _channels = new(StringComparer.Ordinal); + + /// Initializes a new instance of the class. + /// Config-source options; supplies the endpoints, shared key, and deadline. + /// Logger. + /// + /// Seam for tests: builds the generated client for an endpoint. Defaults to a real h2c + /// (one cached per endpoint). + /// + public GrpcDeploymentArtifactFetcher( + IOptions options, + ILogger log, + Func? clientFactory = null) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(log); + + _options = options.Value; + _log = log; + _clientFactory = clientFactory ?? DefaultClientFactory; + } + + /// + public async Task 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 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); + } + + /// + public void Dispose() + { + foreach (var channel in _channels.Values) + channel.Dispose(); + _channels.Clear(); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/IDeploymentArtifactFetcher.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/IDeploymentArtifactFetcher.cs new file mode 100644 index 00000000..c2233af3 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/IDeploymentArtifactFetcher.cs @@ -0,0 +1,28 @@ +namespace ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; + +/// +/// Fetches the deployed-configuration artifact bytes from central over gRPC (per-cluster mesh +/// Phase 3, FetchAndCache mode). The node-side counterpart of central's +/// DeploymentArtifactService. +/// +public interface IDeploymentArtifactFetcher +{ + /// + /// Fetches and reassembles the artifact for from the first + /// configured central endpoint that answers, verifying that + /// SHA-256(reassembled bytes) (lowercase hex) equals + /// — the hash already carried in the dispatch. + /// + /// The deployment id to fetch. + /// + /// The lowercase-hex SHA-256 the reassembled bytes must match (Deployment.RevisionHash). + /// + /// Cancellation token. + /// + /// The verified artifact bytes, or on ANY failure — every configured + /// endpoint unreachable/NotFound, a zero-length stream, or a SHA-256 mismatch. A + /// is "no answer," NEVER an empty configuration: the caller treats it as + /// an apply failure and keeps last-known-good (the #485 contract). + /// + Task FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ZB.MOM.WW.OtOpcUa.Runtime.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ZB.MOM.WW.OtOpcUa.Runtime.csproj index dc524c84..92445c84 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ZB.MOM.WW.OtOpcUa.Runtime.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ZB.MOM.WW.OtOpcUa.Runtime.csproj @@ -8,6 +8,9 @@ + + diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Deployment/GrpcDeploymentArtifactFetcherTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Deployment/GrpcDeploymentArtifactFetcherTests.cs new file mode 100644 index 00000000..49171652 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Deployment/GrpcDeploymentArtifactFetcherTests.cs @@ -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; + +/// +/// 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. +/// +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 factory) + => new( + Options.Create(new ConfigSourceOptions + { + Mode = ConfigSourceOptions.ModeFetchAndCache, + CentralFetchEndpoints = endpoints, + ApiKey = "k", + FetchTimeoutSeconds = 30, + }), + NullLogger.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(); + 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 SingleClient( + IReadOnlyList chunks) + => _ => new FakeClient(chunks); + + private static IReadOnlyList StreamOf(byte[] blob, int chunkSize) + { + var chunks = new List(); + 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; + } + + /// A generated client double whose Fetch streams canned chunks. + private sealed class FakeClient(IReadOnlyList chunks) + : DeploymentArtifactService.DeploymentArtifactServiceClient + { + public override AsyncServerStreamingCall Fetch(FetchRequest request, CallOptions options) + => new( + new ListStreamReader(chunks), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => [], + () => { }); + } + + /// A generated client double whose Fetch throws (endpoint down / NotFound). + private sealed class ThrowingClient(RpcException toThrow) + : DeploymentArtifactService.DeploymentArtifactServiceClient + { + public override AsyncServerStreamingCall Fetch(FetchRequest request, CallOptions options) + => throw toThrow; + } + + private sealed class ListStreamReader(IReadOnlyList chunks) : IAsyncStreamReader + { + private int _index = -1; + + public ArtifactChunk Current => chunks[_index]; + + public Task MoveNext(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + _index++; + return Task.FromResult(_index < chunks.Count); + } + } +}