dohertj2/lmxopcua
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mesh): central DeploymentArtifactService — stream the sealed artifact, NotFound-hide the rest

Browse Source 
Streams a sealed deployment's ArtifactBlob to a fetching driver node in
128-KiB chunks (matching the LocalDb cache chunk size). Unknown id, a
non-sealed deployment, and a zero-length blob all collapse to one
indistinguishable NotFound — existence-hiding plus the #485 serve-side guard
(never stream empty bytes as a valid empty config). The impl is named
DeploymentArtifactGrpcService to avoid colliding with the generated
DeploymentArtifactService container type.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-22 19:08:20 -04:00
parent 62e8f54f71
commit 7d7de482b7
2 changed files with 251 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/DeploymentArtifactService.cs
+92
View File
@@ -0,0 +1,92 @@
using Google.Protobuf;
using Grpc.Core;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.DeploymentArtifact.V1;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
// The generated service container is itself named DeploymentArtifactService; alias its nested server
// base so this impl can keep the natural name without colliding with the generated type.
using GeneratedServiceBase =
ZB.MOM.WW.OtOpcUa.Commons.Protos.DeploymentArtifact.V1.DeploymentArtifactService.DeploymentArtifactServiceBase;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Grpc;
/// <summary>
/// Central-side artifact serve (per-cluster mesh Phase 3). Streams a sealed deployment's
/// <c>ArtifactBlob</c> to a driver node fetching in <c>FetchAndCache</c> mode. Registered only on
/// admin-role nodes (which hold the central SQL connection) and only when
/// <c>ConfigServe:GrpcListenPort &gt; 0</c>.
/// </summary>
/// <remarks>
/// <para>
/// <b>NotFound is deliberately indistinguishable across three cases</b> — unknown id, a
/// deployment that is not sealed, and an empty blob. The first is existence-hiding; the last
/// is the #485 rule (a zero-length artifact is "no answer," never "a configuration with no
/// drivers"). Streaming empty bytes as if valid is exactly what that defect class forbids.
/// </para>
/// </remarks>
public sealed class DeploymentArtifactGrpcService : GeneratedServiceBase
{
/// <summary>Chunk size, matching <c>LocalDbDeploymentArtifactCache.ChunkSize</c> (128 KiB).</summary>
private const int ChunkSize = 128 * 1024;
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
private readonly ILogger<DeploymentArtifactGrpcService> _log;
/// <summary>Initializes a new instance of the <see cref="DeploymentArtifactGrpcService"/> class.</summary>
/// <param name="dbFactory">Factory for the central config database holding the artifact bytes.</param>
/// <param name="log">Logger.</param>
public DeploymentArtifactGrpcService(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
ILogger<DeploymentArtifactGrpcService> log)
{
_dbFactory = dbFactory;
_log = log;
}
/// <inheritdoc />
public override async Task Fetch(
FetchRequest request,
IServerStreamWriter<ArtifactChunk> responseStream,
ServerCallContext context)
{
ArgumentNullException.ThrowIfNull(request);
ArgumentNullException.ThrowIfNull(responseStream);
ArgumentNullException.ThrowIfNull(context);
if (!Guid.TryParse(request.DeploymentId, out var id))
throw new RpcException(new Status(StatusCode.NotFound, "unknown deployment"));
await using var db = await _dbFactory.CreateDbContextAsync(context.CancellationToken);
var row = await db.Deployments
.AsNoTracking()
.Where(d => d.DeploymentId == id)
.Select(d => new { d.Status, d.ArtifactBlob })
.FirstOrDefaultAsync(context.CancellationToken);
// Existence-hiding + #485: unknown / not-sealed / empty collapse to one NotFound.
if (row is null || row.Status != DeploymentStatus.Sealed || row.ArtifactBlob.Length == 0)
{
_log.LogDebug(
"Artifact fetch for {DeploymentId} -> NotFound (found={Found}, sealed={Sealed}, bytes={Bytes})",
request.DeploymentId, row is not null,
row is not null && row.Status == DeploymentStatus.Sealed,
row?.ArtifactBlob.Length ?? 0);
throw new RpcException(new Status(StatusCode.NotFound, "unknown deployment"));
}
for (var offset = 0; offset < row.ArtifactBlob.Length; offset += ChunkSize)
{
var len = Math.Min(ChunkSize, row.ArtifactBlob.Length - offset);
await responseStream.WriteAsync(
new ArtifactChunk { Data = ByteString.CopyFrom(row.ArtifactBlob, offset, len) },
context.CancellationToken);
}
_log.LogInformation(
"Served artifact for {DeploymentId} ({Bytes} bytes, {Chunks} chunks)",
request.DeploymentId, row.ArtifactBlob.Length,
(row.ArtifactBlob.Length + ChunkSize - 1) / ChunkSize);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/DeploymentArtifactServiceTests.cs
+159
View File
@@ -0,0 +1,159 @@
using Google.Protobuf;
using Grpc.Core;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.AdminUI.Grpc;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.DeploymentArtifact.V1;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Grpc;
/// <summary>
/// The serve side of Phase 3: streams a sealed deployment's bytes, and collapses unknown /
/// not-sealed / empty into one indistinguishable NotFound (existence-hiding + #485).
/// </summary>
public class DeploymentArtifactServiceTests
{
private static IDbContextFactory<OtOpcUaConfigDbContext> Factory(string name) => new NamedFactory(name);
private static DeploymentArtifactGrpcService Service(string dbName) =>
new(Factory(dbName), NullLogger<DeploymentArtifactGrpcService>.Instance);
private static Guid Seed(string dbName, DeploymentStatus status, byte[] blob)
{
var id = Guid.NewGuid();
using var db = new OtOpcUaConfigDbContext(
new DbContextOptionsBuilder<OtOpcUaConfigDbContext>().UseInMemoryDatabase(dbName).Options);
db.Deployments.Add(new Deployment
{
DeploymentId = id,
RevisionHash = new string('a', 64),
Status = status,
CreatedBy = "test",
ArtifactBlob = blob,
});
db.SaveChanges();
return id;
}
[Fact]
public async Task A_sealed_non_empty_blob_streams_back_byte_equal_and_chunk_bounded()
{
var dbName = $"artsvc-{Guid.NewGuid():N}";
// ~300 KB > 2 * 128 KiB -> exactly 3 chunks.
var blob = new byte[300 * 1024];
for (var i = 0; i < blob.Length; i++) blob[i] = (byte)(i % 251);
var id = Seed(dbName, DeploymentStatus.Sealed, blob);
var writer = new CollectingStreamWriter();
await Service(dbName).Fetch(
new FetchRequest { DeploymentId = id.ToString() }, writer, new FakeServerCallContext());
writer.Chunks.Count.ShouldBe(3);
writer.Chunks[0].Data.Length.ShouldBe(128 * 1024);
writer.Chunks[1].Data.Length.ShouldBe(128 * 1024);
writer.Reassembled().ShouldBe(blob);
}
[Fact]
public async Task An_unknown_id_throws_NotFound()
{
var dbName = $"artsvc-{Guid.NewGuid():N}";
Seed(dbName, DeploymentStatus.Sealed, [1, 2, 3]);
var ex = await Should.ThrowAsync<RpcException>(() => Service(dbName).Fetch(
new FetchRequest { DeploymentId = Guid.NewGuid().ToString() },
new CollectingStreamWriter(), new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.NotFound);
}
[Fact]
public async Task A_non_sealed_deployment_throws_NotFound()
{
var dbName = $"artsvc-{Guid.NewGuid():N}";
var id = Seed(dbName, DeploymentStatus.AwaitingApplyAcks, [1, 2, 3]);
var ex = await Should.ThrowAsync<RpcException>(() => Service(dbName).Fetch(
new FetchRequest { DeploymentId = id.ToString() },
new CollectingStreamWriter(), new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.NotFound);
}
[Fact]
public async Task A_zero_length_blob_throws_NotFound()
{
// The serve-side #485 guard: never stream empty bytes as if they were a valid empty config.
var dbName = $"artsvc-{Guid.NewGuid():N}";
var id = Seed(dbName, DeploymentStatus.Sealed, []);
var ex = await Should.ThrowAsync<RpcException>(() => Service(dbName).Fetch(
new FetchRequest { DeploymentId = id.ToString() },
new CollectingStreamWriter(), new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.NotFound);
}
[Fact]
public async Task A_non_guid_id_throws_NotFound()
{
var ex = await Should.ThrowAsync<RpcException>(() => Service($"artsvc-{Guid.NewGuid():N}").Fetch(
new FetchRequest { DeploymentId = "not-a-guid" },
new CollectingStreamWriter(), new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.NotFound);
}
private sealed class CollectingStreamWriter : IServerStreamWriter<ArtifactChunk>
{
public List<ArtifactChunk> Chunks { get; } = [];
public WriteOptions? WriteOptions { get; set; }
public Task WriteAsync(ArtifactChunk message)
{
Chunks.Add(message);
return Task.CompletedTask;
}
public byte[] Reassembled()
{
using var ms = new MemoryStream();
foreach (var c in Chunks) c.Data.WriteTo(ms);
return ms.ToArray();
}
}
private sealed class NamedFactory(string name) : IDbContextFactory<OtOpcUaConfigDbContext>
{
public OtOpcUaConfigDbContext CreateDbContext() =>
new(new DbContextOptionsBuilder<OtOpcUaConfigDbContext>().UseInMemoryDatabase(name).Options);
public Task<OtOpcUaConfigDbContext> CreateDbContextAsync(CancellationToken ct = default) =>
Task.FromResult(CreateDbContext());
}
/// <summary>Minimal <see cref="ServerCallContext"/> — the service reads only the cancellation token.</summary>
private sealed class FakeServerCallContext : ServerCallContext
{
protected override string MethodCore => "/deployment_artifact.v1.DeploymentArtifactService/Fetch";
protected override string HostCore => "localhost";
protected override string PeerCore => "ipv4:127.0.0.1:0";
protected override DateTime DeadlineCore => DateTime.MaxValue;
protected override Metadata RequestHeadersCore => [];
protected override CancellationToken CancellationTokenCore => CancellationToken.None;
protected override Metadata ResponseTrailersCore { get; } = [];
protected override Status StatusCore { get; set; }
protected override WriteOptions? WriteOptionsCore { get; set; }
protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());
protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
throw new NotSupportedException();
protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
}
}