diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/KestrelHttpBinding.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/KestrelHttpBinding.cs new file mode 100644 index 00000000..43b52de1 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/KestrelHttpBinding.cs @@ -0,0 +1,75 @@ +using System.Net; +using Microsoft.AspNetCore.Server.Kestrel.Core; + +namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; + +/// +/// One HTTP endpoint the host was asked to serve, parsed out of the configured URL list so it +/// can be re-bound explicitly. +/// +/// The host component as written — +, *, a hostname, or an IP. +/// The port. +/// True when the URL used the https scheme. +public sealed record KestrelHttpBinding(string Host, int Port, bool IsSecure) +{ + /// + /// Parses a semicolon-separated URL list (the ASPNETCORE_URLS / urls format) + /// into bindings. Unparseable entries are skipped rather than throwing — a malformed URL + /// must not take the process down at startup. + /// + /// The configured URL list, or null/empty when none is configured. + /// The parsed bindings, in the order given; empty when nothing is configured. + public static IReadOnlyList Parse(string? urls) + { + if (string.IsNullOrWhiteSpace(urls)) + return []; + + var result = new List(); + foreach (var raw in urls.Split(';', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)) + { + // Uri cannot parse the "+"/"*" wildcard hosts Kestrel accepts, so substitute a + // placeholder host purely to get scheme/port out, then keep the original host token. + var isWildcard = raw.Contains("://+", StringComparison.Ordinal) + || raw.Contains("://*", StringComparison.Ordinal); + var probe = isWildcard + ? raw.Replace("://+", "://placeholder", StringComparison.Ordinal) + .Replace("://*", "://placeholder", StringComparison.Ordinal) + : raw; + + if (!Uri.TryCreate(probe, UriKind.Absolute, out var uri)) + continue; + + var host = isWildcard + ? raw.Contains("://+", StringComparison.Ordinal) ? "+" : "*" + : uri.Host; + + result.Add(new KestrelHttpBinding(host, uri.Port, uri.Scheme == Uri.UriSchemeHttps)); + } + + return result; + } + + /// + /// Applies this binding to Kestrel, choosing the narrowest listen call the host component + /// allows. + /// + /// + /// A hostname that is neither a literal IP nor localhost cannot be reliably mapped to + /// a local interface, so it falls back to — + /// the safe superset. Narrowing it and guessing wrong would silently stop serving. + /// + /// The Kestrel options being configured. + public void Apply(KestrelServerOptions options) + { + ArgumentNullException.ThrowIfNull(options); + + if (Host is "+" or "*") + options.ListenAnyIP(Port); + else if (IPAddress.TryParse(Host, out var address)) + options.Listen(address, Port); + else if (string.Equals(Host, "localhost", StringComparison.OrdinalIgnoreCase)) + options.ListenLocalhost(Port); + else + options.ListenAnyIP(Port); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 30c36854..1c772c56 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -1,7 +1,9 @@ using Akka.Actor; using Akka.Hosting; +using Microsoft.AspNetCore.Server.Kestrel.Core; using Microsoft.Extensions.DependencyInjection.Extensions; using Serilog; +using ZB.MOM.WW.LocalDb.Replication; using Serilog.Events; using ZB.MOM.WW.OtOpcUa.AdminUI; using ZB.MOM.WW.OtOpcUa.AdminUI.Clients; @@ -379,6 +381,71 @@ builder.Services.AddOtOpcUaSecrets(builder.Configuration); builder.Services.AddOtOpcUaHealth(); builder.Services.AddOtOpcUaObservability(builder.Configuration); +// --------------------------------------------------------------------------------------------- +// LocalDb sync listener (default-OFF). +// +// DANGER: any explicit Kestrel Listen* call makes Kestrel IGNORE ASPNETCORE_URLS/urls ENTIRELY +// (it logs "Overriding address(es)"). The host has no ConfigureKestrel today and binds solely via +// that configuration, so adding a listener naively would silently unbind the AdminUI + deploy API +// behind Traefik. Everything the host was already asked to serve is therefore re-bound explicitly +// in the same block. +// +// The listener is HTTP/2-ONLY on purpose: the sync client speaks prior-knowledge h2c, which a +// cleartext Http1AndHttp2 endpoint cannot negotiate (there is no ALPN without TLS). Hence a +// dedicated port rather than multiplexing onto the main one. +// +// When LocalDb:SyncListenPort is 0 (the default) none of this runs and URL binding is untouched. +// --------------------------------------------------------------------------------------------- +var syncListenPort = LocalDbRegistration.SyncListenPort(builder.Configuration); +if (hasDriver && syncListenPort > 0) +{ + var configuredUrls = builder.Configuration["urls"] + ?? builder.Configuration["ASPNETCORE_URLS"]; + var existingBindings = KestrelHttpBinding.Parse(configuredUrls); + + // A driver-only Windows-service node gets NO ASPNETCORE_URLS (Install-Services.ps1 sets it + // only under $hasAdmin), so "nothing configured" is a real, supported state — not an error. + // Kestrel's own default is localhost:5000; re-state it explicitly, because taking over + // configuration means taking over the default too. Health probes hit this port. + if (existingBindings.Count == 0) + { + existingBindings = [new KestrelHttpBinding("localhost", 5000, IsSecure: false)]; + Log.Information( + "LocalDb sync listener enabled with no URLs configured; re-binding Kestrel's default " + + "http://localhost:5000 alongside the sync port."); + } + + // Re-binding an https endpoint would need its certificate configuration replayed too, which + // this host does not model (TLS is terminated at Traefik). Rather than silently serve it + // without TLS — or drop it — refuse to take over Kestrel at all and leave the existing + // surface exactly as it was. The operator gets a loud, actionable error instead of an + // AdminUI that stopped answering. + if (existingBindings.Any(b => b.IsSecure)) + { + Log.Error( + "LocalDb:SyncListenPort is set to {Port} but this host serves HTTPS endpoint(s) ({Urls}). " + + "Binding the sync listener requires re-binding every existing endpoint explicitly, and the " + + "HTTPS certificate configuration cannot be replayed safely. The sync listener is DISABLED; " + + "terminate TLS upstream (as the docker-dev rig does) or leave replication off on this node.", + syncListenPort, configuredUrls); + syncListenPort = 0; + } + else + { + builder.WebHost.ConfigureKestrel(kestrel => + { + foreach (var binding in existingBindings) + binding.Apply(kestrel); + + kestrel.ListenAnyIP(syncListenPort, o => o.Protocols = HttpProtocols.Http2); + }); + + Log.Information( + "LocalDb sync listener bound on :{SyncPort} (h2c); re-bound existing endpoint(s) {Urls}.", + syncListenPort, configuredUrls ?? "http://localhost:5000 (Kestrel default)"); + } +} + var app = builder.Build(); // AddZbSerilog registers Serilog as the MEL logging provider but does NOT assign the static @@ -409,6 +476,15 @@ if (hasAdmin) app.MapOtOpcUaDeployApi(app.Configuration); } +// Passive LocalDb sync endpoint. Gated on the same port check that bound the listener, so a +// default-OFF or admin-only graph carries no sync surface whatsoever. Mapping it would be safe +// even unauthenticated — LocalDbSyncAuthInterceptor fail-closes with no ApiKey configured — but +// not mapping it at all is the stronger default. +if (hasDriver && syncListenPort > 0) +{ + app.MapZbLocalDbSync(); +} + app.MapOtOpcUaHealth(); app.MapOtOpcUaMetrics(); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/IDeploymentArtifactCache.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/IDeploymentArtifactCache.cs new file mode 100644 index 00000000..32b841c6 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/IDeploymentArtifactCache.cs @@ -0,0 +1,68 @@ +namespace ZB.MOM.WW.OtOpcUa.Runtime.Deployment; + +/// +/// Node-local durable cache of the deployment artifact this node last applied. +/// +/// +/// +/// Exists so a driver node can boot into its last-known-good address space when the central +/// configuration database is unreachable. Without it, a control-plane outage that outlives a +/// node restart leaves that node with no address space at all — the plant loses visibility +/// for a reason that has nothing to do with the plant. +/// +/// +public interface IDeploymentArtifactCache +{ + /// + /// Persists as the cluster's current deployment. + /// + /// + /// + /// Idempotent per : re-storing the same deployment + /// replaces its chunks rather than accumulating orphans. Retention is bounded to the two + /// newest deployments per cluster, so a long-lived node cannot grow its local database + /// without limit. + /// + /// + Task StoreAsync(string clusterId, string deploymentId, string revisionHash, + byte[] artifact, CancellationToken ct = default); + + /// + /// Reads the cached artifact for , or + /// when nothing is cached or the cached bytes fail their integrity check. + /// + /// + /// + /// A failed integrity check is deliberately indistinguishable from a miss. Booting a + /// plant from a truncated address space is strictly worse than booting from none: the + /// missing half looks like a deliberate configuration rather than an error. + /// + /// + Task GetCurrentAsync(string clusterId, CancellationToken ct = default); + + /// + /// Reads the single cached pointer without knowing the cluster id. + /// + /// + /// + /// The boot seam needs this. A node's cluster id is only derivable from an artifact it + /// has already loaded, or from the central database — which is precisely what is + /// unreachable in the scenario this cache exists to survive. So the cold path reads the + /// newest pointer unkeyed. + /// + /// + /// More than one pointer row means the node was re-homed between clusters. The newest + /// wins, but the choice is logged as a warning naming both clusters — a node silently + /// booting a neighbouring cluster's configuration is exactly the failure that must never + /// be quiet. + /// + /// + Task GetCurrentUnkeyedAsync(CancellationToken ct = default); +} + +/// +/// A deployment artifact recovered from the node-local cache, with the metadata needed to decide +/// whether it is still current once the control plane is reachable again. +/// +public sealed record CachedDeploymentArtifact( + string DeploymentId, string RevisionHash, byte[] Artifact, DateTimeOffset AppliedAtUtc); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/LocalDbDeploymentArtifactCache.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/LocalDbDeploymentArtifactCache.cs new file mode 100644 index 00000000..e6b97cc2 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/LocalDbDeploymentArtifactCache.cs @@ -0,0 +1,316 @@ +using System.Globalization; +using System.Security.Cryptography; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.LocalDb; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Deployment; + +/// +/// backed by the replicated LocalDb deployment-cache +/// tables. +/// +/// +/// +/// Why the artifact is stored as base64 chunks. See +/// — replication rejects BLOB columns and batches by row +/// count against gRPC's message cap, so a whole-artifact row could never be delivered. +/// Everything in this class that looks like ceremony (chunk indices, a chunk count, a +/// separate SHA over the raw bytes) exists to make that split safely reversible. +/// +/// +public sealed class LocalDbDeploymentArtifactCache : IDeploymentArtifactCache +{ + /// + /// Raw bytes per chunk, before base64 expansion. + /// + /// + /// 128 KiB raw is ≈ 171 KB encoded, which keeps a worst-case replication batch inside gRPC's + /// 4 MB default cap with room to spare. + /// + private const int ChunkSize = 128 * 1024; + + /// + /// How many deployments per cluster survive a store. + /// + /// + /// Two, not one: the previous artifact is what a rollback would need, and keeping it costs + /// one artifact's worth of disk. Three would only add a generation nobody rolls back to. + /// + private const int RetainedDeployments = 2; + + private readonly ILocalDb _db; + private readonly ILogger _logger; + + public LocalDbDeploymentArtifactCache(ILocalDb db, ILogger logger) + { + ArgumentNullException.ThrowIfNull(db); + ArgumentNullException.ThrowIfNull(logger); + + _db = db; + _logger = logger; + } + + /// + public async Task StoreAsync(string clusterId, string deploymentId, string revisionHash, + byte[] artifact, CancellationToken ct = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(clusterId); + ArgumentException.ThrowIfNullOrWhiteSpace(deploymentId); + ArgumentException.ThrowIfNullOrWhiteSpace(revisionHash); + ArgumentNullException.ThrowIfNull(artifact); + + var sha = Convert.ToHexString(SHA256.HashData(artifact)); + var cachedAtUtc = FormatTimestamp(DateTimeOffset.UtcNow); + var chunkCount = (artifact.Length + ChunkSize - 1) / ChunkSize; + + // One transaction for the whole store. A pointer that commits without its chunks — or + // chunks that commit without the pointer — is a cache that reads back as a corrupt hit + // rather than a clean miss, which is the one outcome this type exists to prevent. + await using var tx = await _db.BeginTransactionAsync(ct); + + // Delete-then-insert rather than upsert-per-chunk: a re-store with FEWER chunks than last + // time would otherwise leave the old tail behind, and those orphans read back as a + // chunk-count mismatch forever. + await tx.ExecuteAsync( + "DELETE FROM deployment_artifacts WHERE deployment_id = @DeploymentId", + new { DeploymentId = deploymentId }, + ct); + + for (var index = 0; index < chunkCount; index++) + { + var offset = index * ChunkSize; + var length = Math.Min(ChunkSize, artifact.Length - offset); + + await tx.ExecuteAsync( + """ + INSERT INTO deployment_artifacts + (deployment_id, chunk_index, cluster_id, revision_hash, chunk_count, + chunk_base64, cached_at_utc) + VALUES + (@DeploymentId, @ChunkIndex, @ClusterId, @RevisionHash, @ChunkCount, + @ChunkBase64, @CachedAtUtc) + """, + new + { + DeploymentId = deploymentId, + ChunkIndex = index, + ClusterId = clusterId, + RevisionHash = revisionHash, + ChunkCount = chunkCount, + ChunkBase64 = Convert.ToBase64String(artifact, offset, length), + CachedAtUtc = cachedAtUtc, + }, + ct); + } + + await tx.ExecuteAsync( + """ + INSERT INTO deployment_pointer + (cluster_id, deployment_id, revision_hash, artifact_sha256, applied_at_utc) + VALUES + (@ClusterId, @DeploymentId, @RevisionHash, @Sha, @AppliedAtUtc) + ON CONFLICT(cluster_id) DO UPDATE SET + deployment_id = excluded.deployment_id, + revision_hash = excluded.revision_hash, + artifact_sha256 = excluded.artifact_sha256, + applied_at_utc = excluded.applied_at_utc + """, + new + { + ClusterId = clusterId, + DeploymentId = deploymentId, + RevisionHash = revisionHash, + Sha = sha, + AppliedAtUtc = cachedAtUtc, + }, + ct); + + // Prune inside the same transaction so the cache is never briefly unbounded. Ordering by + // deployment_id as a tiebreak keeps the survivor set deterministic when two stores land in + // the same timestamp tick. + await tx.ExecuteAsync( + $""" + DELETE FROM deployment_artifacts + WHERE cluster_id = @ClusterId + AND deployment_id NOT IN ( + SELECT deployment_id + FROM deployment_artifacts + WHERE cluster_id = @ClusterId + GROUP BY deployment_id + ORDER BY MAX(cached_at_utc) DESC, deployment_id DESC + LIMIT {RetainedDeployments} + ) + """, + new { ClusterId = clusterId }, + ct); + + await tx.CommitAsync(ct); + } + + /// + public async Task GetCurrentAsync( + string clusterId, CancellationToken ct = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(clusterId); + + var pointers = await _db.QueryAsync( + """ + SELECT deployment_id, revision_hash, artifact_sha256, applied_at_utc + FROM deployment_pointer + WHERE cluster_id = @ClusterId + """, + ReadPointer, + new { ClusterId = clusterId }, + ct); + + if (pointers.Count == 0) + { + _logger.LogDebug("No cached deployment pointer for cluster {ClusterId}.", clusterId); + return null; + } + + return await ReassembleAsync(clusterId, pointers[0], ct); + } + + /// + public async Task GetCurrentUnkeyedAsync(CancellationToken ct = default) + { + // applied_at_utc is round-trip ISO-8601 UTC, so ordering it as TEXT is chronological — + // that is the reason the format is pinned rather than left to the current culture. + var pointers = await _db.QueryAsync( + """ + SELECT cluster_id, deployment_id, revision_hash, artifact_sha256, applied_at_utc + FROM deployment_pointer + ORDER BY applied_at_utc DESC + """, + r => (ClusterId: r.GetString(0), Pointer: new PointerRow( + DeploymentId: r.GetString(1), + RevisionHash: r.GetString(2), + ArtifactSha256: r.GetString(3), + AppliedAtUtc: r.GetString(4))), + parameters: null, + ct); + + if (pointers.Count == 0) + { + _logger.LogDebug("No cached deployment pointer of any cluster."); + return null; + } + + if (pointers.Count > 1) + { + // The node was re-homed between clusters. Booting the newest is the only defensible + // choice, but it must never be a silent one — a wrong-cluster address space presents as + // a plausible configuration, so the operator needs the cluster names to spot it. + _logger.LogWarning( + "Local deployment cache holds pointers for {PointerCount} clusters ({ClusterIds}); " + + "booting the newest ({ChosenClusterId}). This node appears to have been re-homed — " + + "clear the stale cache if that is unexpected.", + pointers.Count, + string.Join(", ", pointers.Select(p => p.ClusterId)), + pointers[0].ClusterId); + } + + return await ReassembleAsync(pointers[0].ClusterId, pointers[0].Pointer, ct); + } + + /// + /// Rebuilds the artifact the pointer names, returning unless every + /// integrity check passes. + /// + private async Task ReassembleAsync( + string clusterId, PointerRow pointer, CancellationToken ct) + { + var chunks = await _db.QueryAsync( + """ + SELECT chunk_count, chunk_base64 + FROM deployment_artifacts + WHERE deployment_id = @DeploymentId + ORDER BY chunk_index + """, + r => (ChunkCount: r.GetInt32(0), Base64: r.GetString(1)), + new { pointer.DeploymentId }, + ct); + + // The chunk_count carried on every row is what makes a partial replica detectable. Without + // it a missing tail chunk is indistinguishable from a shorter artifact. + var expectedChunkCount = chunks.Count > 0 ? chunks[0].ChunkCount : 0; + + if (chunks.Count != expectedChunkCount) + { + _logger.LogWarning( + "Cached deployment {DeploymentId} for cluster {ClusterId} is incomplete: found " + + "{FoundChunks} of {ExpectedChunks} chunks. Treating as a cache miss.", + pointer.DeploymentId, clusterId, chunks.Count, expectedChunkCount); + return null; + } + + byte[] artifact; + try + { + artifact = Decode(chunks.Select(c => c.Base64)); + } + catch (FormatException ex) + { + _logger.LogWarning( + ex, + "Cached deployment {DeploymentId} for cluster {ClusterId} has a chunk that is not " + + "valid base64. Treating as a cache miss.", + pointer.DeploymentId, clusterId); + return null; + } + + var actualSha = Convert.ToHexString(SHA256.HashData(artifact)); + if (!string.Equals(actualSha, pointer.ArtifactSha256, StringComparison.OrdinalIgnoreCase)) + { + _logger.LogWarning( + "Cached deployment {DeploymentId} for cluster {ClusterId} failed its SHA-256 check " + + "(expected {ExpectedSha}, computed {ActualSha}). Treating as a cache miss.", + pointer.DeploymentId, clusterId, pointer.ArtifactSha256, actualSha); + return null; + } + + if (!DateTimeOffset.TryParse(pointer.AppliedAtUtc, CultureInfo.InvariantCulture, + DateTimeStyles.RoundtripKind, out var appliedAtUtc)) + { + _logger.LogWarning( + "Cached deployment {DeploymentId} for cluster {ClusterId} has an unparseable " + + "applied_at_utc ({AppliedAtUtc}). Treating as a cache miss.", + pointer.DeploymentId, clusterId, pointer.AppliedAtUtc); + return null; + } + + return new CachedDeploymentArtifact( + pointer.DeploymentId, pointer.RevisionHash, artifact, appliedAtUtc); + } + + private static byte[] Decode(IEnumerable chunksInOrder) + { + using var buffer = new MemoryStream(); + + foreach (var chunk in chunksInOrder) + { + var bytes = Convert.FromBase64String(chunk); + buffer.Write(bytes, 0, bytes.Length); + } + + return buffer.ToArray(); + } + + private static PointerRow ReadPointer(Microsoft.Data.Sqlite.SqliteDataReader reader) + => new( + DeploymentId: reader.GetString(0), + RevisionHash: reader.GetString(1), + ArtifactSha256: reader.GetString(2), + AppliedAtUtc: reader.GetString(3)); + + /// + /// Round-trip ("O") UTC, so lexicographic ordering of the stored TEXT is chronological + /// ordering — the retention query sorts on it directly. + /// + private static string FormatTimestamp(DateTimeOffset value) + => value.ToUniversalTime().ToString("O", CultureInfo.InvariantCulture); + + private sealed record PointerRow( + string DeploymentId, string RevisionHash, string ArtifactSha256, string AppliedAtUtc); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSyncListenerTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSyncListenerTests.cs new file mode 100644 index 00000000..0a1c69cc --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSyncListenerTests.cs @@ -0,0 +1,197 @@ +using System.Net; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; + +/// +/// LocalDb Phase 1 (Task 5) — the dedicated h2c sync listener and, more importantly, the +/// re-binding of the endpoints the host was already serving. +/// +/// +/// +/// Why this file exists. The host binds exclusively via ASPNETCORE_URLS and has +/// no ConfigureKestrel call. Any explicit Listen* makes Kestrel discard that +/// configuration wholesale — it logs "Overriding address(es)" and serves only what was +/// listed explicitly. Getting this wrong unbinds the AdminUI and the deploy API behind +/// Traefik, and it fails silently: the process starts, logs look normal, and nothing answers. +/// +/// +/// These tests drive a minimal WebApplication shaped exactly like Program.cs's +/// block rather than booting the real host, which needs SQL Server, an Akka mesh and LDAP. +/// What is under test is the Kestrel binding contract, and that is fully reproduced here. +/// +/// +public sealed class LocalDbSyncListenerTests +{ + // --------------------------------------------------------------------------------------- + // KestrelHttpBinding.Parse + // --------------------------------------------------------------------------------------- + + [Fact] + public void Parse_NullOrEmpty_YieldsNoBindings() + { + // "Nothing configured" is a real supported state: Install-Services.ps1 sets ASPNETCORE_URLS + // only for admin nodes, so a driver-only service genuinely has none. + KestrelHttpBinding.Parse(null).ShouldBeEmpty(); + KestrelHttpBinding.Parse("").ShouldBeEmpty(); + KestrelHttpBinding.Parse(" ").ShouldBeEmpty(); + } + + [Theory] + [InlineData("http://+:9000", "+", 9000)] + [InlineData("http://*:9000", "*", 9000)] + [InlineData("http://localhost:9000", "localhost", 9000)] + [InlineData("http://0.0.0.0:8080", "0.0.0.0", 8080)] + [InlineData("http://127.0.0.1:5001", "127.0.0.1", 5001)] + public void Parse_SingleUrl_ExtractsHostAndPort(string url, string expectedHost, int expectedPort) + { + // The "+" and "*" wildcards are Kestrel-isms that System.Uri cannot parse — the docker-dev + // rig uses "http://+:9000", so mishandling them would break every rig node. + var bindings = KestrelHttpBinding.Parse(url); + + bindings.Count.ShouldBe(1); + bindings[0].Host.ShouldBe(expectedHost); + bindings[0].Port.ShouldBe(expectedPort); + bindings[0].IsSecure.ShouldBeFalse(); + } + + [Fact] + public void Parse_MultipleUrls_PreservesAllOfThem() + { + // Dropping one of several configured endpoints is the exact silent-unbind failure this + // whole mechanism exists to avoid. + var bindings = KestrelHttpBinding.Parse("http://+:9000;http://localhost:9100"); + + bindings.Count.ShouldBe(2); + bindings[0].Port.ShouldBe(9000); + bindings[1].Port.ShouldBe(9100); + } + + [Fact] + public void Parse_HttpsUrl_IsFlaggedSecure() + { + // Program.cs refuses to take over Kestrel when any endpoint is HTTPS, because replaying + // certificate configuration is not modelled here. This flag is what drives that refusal. + KestrelHttpBinding.Parse("https://+:443")[0].IsSecure.ShouldBeTrue(); + } + + [Fact] + public void Parse_MalformedEntry_IsSkipped_NotThrown() + { + // A typo'd URL must not take the process down at startup. + KestrelHttpBinding.Parse("not a url;http://+:9000").ShouldHaveSingleItem().Port.ShouldBe(9000); + } + + // --------------------------------------------------------------------------------------- + // The real Kestrel contract + // --------------------------------------------------------------------------------------- + + [Fact] + public async Task ExplicitListen_ReBindsTheExistingSurface_AndAddsAnH2cListener() + { + // THE test. Both ports must answer simultaneously: the re-bound HTTP/1.1 surface (proving + // the ASPNETCORE_URLS override was compensated for) and the HTTP/2-only sync port (proving + // prior-knowledge h2c works, which a cleartext Http1AndHttp2 endpoint cannot do). + var httpPort = GetFreePort(); + var syncPort = GetFreePort(); + + var builder = WebApplication.CreateBuilder(); + builder.Environment.ApplicationName = typeof(LocalDbSyncListenerTests).Assembly.GetName().Name!; + builder.Services.AddGrpc(); + + // Exactly the shape Program.cs applies. + foreach (var binding in KestrelHttpBinding.Parse($"http://localhost:{httpPort}")) + { + var captured = binding; + builder.WebHost.ConfigureKestrel(k => captured.Apply(k)); + } + + builder.WebHost.ConfigureKestrel(k => + k.ListenAnyIP(syncPort, o => o.Protocols = HttpProtocols.Http2)); + + await using var app = builder.Build(); + app.MapGet("/healthz", () => "ok"); + await app.StartAsync(TestContext.Current.CancellationToken); + + try + { + using var http1 = new HttpClient(); + var body = await http1.GetStringAsync( + $"http://localhost:{httpPort}/healthz", TestContext.Current.CancellationToken); + body.ShouldBe("ok"); + + // Prior-knowledge h2c against the sync port. A 404 is a perfectly good result — it + // proves the HTTP/2 connection was established and the request was routed, which is + // the only thing in question here. + using var http2 = new HttpClient + { + DefaultRequestVersion = HttpVersion.Version20, + DefaultVersionPolicy = HttpVersionPolicy.RequestVersionExact, + }; + using var response = await http2.GetAsync( + $"http://localhost:{syncPort}/", TestContext.Current.CancellationToken); + + response.Version.ShouldBe(HttpVersion.Version20); + } + finally + { + await app.StopAsync(TestContext.Current.CancellationToken); + } + } + + [Fact] + public async Task ExplicitListen_WithoutReBinding_SilentlyDiscardsTheConfiguredUrls() + { + // POSITIVE CONTROL for the whole task. If Kestrel did NOT override configured URLs, the + // re-binding above would be pointless ceremony. This pins the actual behaviour: a single + // explicit Listen* call makes the ASPNETCORE_URLS port stop answering entirely — no + // exception, no failed startup, just silence. That is the regression being guarded against. + var configuredPort = GetFreePort(); + var explicitPort = GetFreePort(); + + var builder = WebApplication.CreateBuilder(); + builder.Environment.ApplicationName = typeof(LocalDbSyncListenerTests).Assembly.GetName().Name!; + builder.WebHost.UseUrls($"http://localhost:{configuredPort}"); + + // Deliberately NOT re-binding configuredPort — this is the mistake being demonstrated. + builder.WebHost.ConfigureKestrel(k => k.ListenAnyIP(explicitPort)); + + await using var app = builder.Build(); + app.MapGet("/healthz", () => "ok"); + await app.StartAsync(TestContext.Current.CancellationToken); + + try + { + using var client = new HttpClient { Timeout = TimeSpan.FromSeconds(5) }; + + // The explicitly listed port serves. + (await client.GetStringAsync( + $"http://localhost:{explicitPort}/healthz", + TestContext.Current.CancellationToken)).ShouldBe("ok"); + + // The configured one does not — nothing is listening there at all. + await Should.ThrowAsync(() => client.GetStringAsync( + $"http://localhost:{configuredPort}/healthz", + TestContext.Current.CancellationToken)); + } + finally + { + await app.StopAsync(TestContext.Current.CancellationToken); + } + } + + private static int GetFreePort() + { + using var listener = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Deployment/LocalDbDeploymentArtifactCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Deployment/LocalDbDeploymentArtifactCacheTests.cs new file mode 100644 index 00000000..a2f34b20 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Deployment/LocalDbDeploymentArtifactCacheTests.cs @@ -0,0 +1,254 @@ +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.OtOpcUa.Runtime.Deployment; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Deployment; + +/// +/// LocalDb Phase 1 (Task 6) — the chunked deployment-artifact cache. +/// +/// +/// +/// Driven against a real temp-file rather than a fake. The behaviours +/// under test — chunk reassembly order, SHA-256 integrity, retention pruning, upsert +/// semantics — live entirely in SQL, so a mocked seam would assert nothing about them. +/// +/// +/// This project cannot reference the Host, so the schema setup here mirrors +/// LocalDbSetup.OnReady rather than calling it. The load-bearing +/// DDL → RegisterReplicated ordering is pinned by the Host's own +/// LocalDbSetupTests; what matters here is only that the tables exist. +/// +/// +public sealed class LocalDbDeploymentArtifactCacheTests : IDisposable +{ + private const string ClusterA = "SITE-A"; + private const string ClusterB = "SITE-B"; + + private readonly string _dbPath = + Path.Combine(Path.GetTempPath(), $"otopcua-artifact-cache-{Guid.NewGuid():N}.db"); + + private readonly ServiceProvider _provider; + private readonly ILocalDb _db; + + public LocalDbDeploymentArtifactCacheTests() + { + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = _dbPath }) + .Build(); + + _provider = new ServiceCollection() + .AddZbLocalDb(configuration, db => + { + using (var connection = db.CreateConnection()) + { + DeploymentCacheSchema.Apply(connection); + } + + db.RegisterReplicated(DeploymentCacheSchema.ArtifactsTable); + db.RegisterReplicated(DeploymentCacheSchema.PointerTable); + }) + .BuildServiceProvider(); + + _db = _provider.GetRequiredService(); + } + + private LocalDbDeploymentArtifactCache CreateCache( + ILogger? logger = null) + => new(_db, logger ?? NullLogger.Instance); + + private static byte[] RandomArtifact(int length) + { + var bytes = new byte[length]; + Random.Shared.NextBytes(bytes); + return bytes; + } + + private async Task CountChunksAsync(string? deploymentId = null) + { + var sql = deploymentId is null + ? "SELECT COUNT(*) FROM deployment_artifacts" + : "SELECT COUNT(*) FROM deployment_artifacts WHERE deployment_id = @DeploymentId"; + + var rows = await _db.QueryAsync(sql, r => r.GetInt32(0), + deploymentId is null ? null : new { DeploymentId = deploymentId }); + + return rows[0]; + } + + private async Task> DistinctDeploymentIdsAsync() + => await _db.QueryAsync( + "SELECT DISTINCT deployment_id FROM deployment_artifacts ORDER BY deployment_id", + r => r.GetString(0)); + + [Fact] + public async Task StoreThenGetCurrent_RoundTripsAMultiChunkArtifactByteForByte() + { + // 300 KiB against a 128 KiB chunk forces three chunks, so this covers the reassembly + // ordering that a single-chunk artifact would never exercise. + var artifact = RandomArtifact(300 * 1024); + var cache = CreateCache(); + + await cache.StoreAsync(ClusterA, "dep-1", "rev-1", artifact); + + var cached = await cache.GetCurrentAsync(ClusterA); + + cached.ShouldNotBeNull(); + cached.DeploymentId.ShouldBe("dep-1"); + cached.RevisionHash.ShouldBe("rev-1"); + cached.Artifact.ShouldBe(artifact); + (await CountChunksAsync("dep-1")).ShouldBe(3); + } + + [Fact] + public async Task Store_KeepsOnlyTheTwoNewestDeploymentsPerCluster() + { + var cache = CreateCache(); + + await cache.StoreAsync(ClusterA, "dep-1", "rev-1", RandomArtifact(64)); + await cache.StoreAsync(ClusterA, "dep-2", "rev-2", RandomArtifact(64)); + await cache.StoreAsync(ClusterA, "dep-3", "rev-3", RandomArtifact(64)); + + // Bounded retention is what stops a node that redeploys daily from filling its disk with + // address spaces nobody will ever roll back to. + (await DistinctDeploymentIdsAsync()).ShouldBe(["dep-2", "dep-3"]); + + var cached = await cache.GetCurrentAsync(ClusterA); + cached.ShouldNotBeNull(); + cached.DeploymentId.ShouldBe("dep-3"); + } + + [Fact] + public async Task GetCurrent_ReturnsNullWhenAChunkIsCorrupt() + { + var cache = CreateCache(); + await cache.StoreAsync(ClusterA, "dep-1", "rev-1", RandomArtifact(300 * 1024)); + + // Valid base64 of the wrong bytes: this passes decoding and the chunk-count check, so only + // the SHA-256 comparison can catch it. That is the check being pinned. + await _db.ExecuteAsync( + "UPDATE deployment_artifacts SET chunk_base64 = @Chunk WHERE deployment_id = @DeploymentId AND chunk_index = 1", + new { Chunk = Convert.ToBase64String(RandomArtifact(128 * 1024)), DeploymentId = "dep-1" }); + + (await cache.GetCurrentAsync(ClusterA)).ShouldBeNull(); + } + + [Fact] + public async Task GetCurrent_ReturnsNullWhenAChunkIsMissing() + { + var cache = CreateCache(); + await cache.StoreAsync(ClusterA, "dep-1", "rev-1", RandomArtifact(300 * 1024)); + + // A partially replicated artifact is the realistic version of this: the pointer row arrives + // before the last chunk does. Reassembling what is present would yield a plausible-looking + // but silently truncated address space. + await _db.ExecuteAsync( + "DELETE FROM deployment_artifacts WHERE deployment_id = @DeploymentId AND chunk_index = 2", + new { DeploymentId = "dep-1" }); + + (await cache.GetCurrentAsync(ClusterA)).ShouldBeNull(); + } + + [Fact] + public async Task GetCurrent_ReturnsNullWhenNoPointerExists() + { + var cache = CreateCache(); + + (await cache.GetCurrentAsync(ClusterA)).ShouldBeNull(); + } + + [Fact] + public async Task Store_IsIdempotentForTheSameDeploymentId() + { + var cache = CreateCache(); + + await cache.StoreAsync(ClusterA, "dep-1", "rev-1", RandomArtifact(300 * 1024)); + + // The re-store is smaller, so a delete-before-insert that did not happen would leave the + // tail chunks of the first artifact behind — orphans that also break the chunk-count check. + var second = RandomArtifact(200 * 1024); + await cache.StoreAsync(ClusterA, "dep-1", "rev-1b", second); + + (await CountChunksAsync("dep-1")).ShouldBe(2); + (await CountChunksAsync()).ShouldBe(2); + + var cached = await cache.GetCurrentAsync(ClusterA); + cached.ShouldNotBeNull(); + cached.RevisionHash.ShouldBe("rev-1b"); + cached.Artifact.ShouldBe(second); + } + + [Fact] + public async Task GetCurrentUnkeyed_ReturnsTheOnlyCachedArtifact() + { + var artifact = RandomArtifact(300 * 1024); + var cache = CreateCache(); + + await cache.StoreAsync(ClusterA, "dep-1", "rev-1", artifact); + + var cached = await cache.GetCurrentUnkeyedAsync(); + + cached.ShouldNotBeNull(); + cached.DeploymentId.ShouldBe("dep-1"); + cached.Artifact.ShouldBe(artifact); + } + + [Fact] + public async Task GetCurrentUnkeyed_TakesTheNewestPointerAndWarnsNamingBothClusters() + { + var newest = RandomArtifact(1024); + var logger = new CapturingLogger(); + var cache = CreateCache(logger); + + await cache.StoreAsync(ClusterA, "dep-a", "rev-a", RandomArtifact(1024)); + await cache.StoreAsync(ClusterB, "dep-b", "rev-b", newest); + + var cached = await cache.GetCurrentUnkeyedAsync(); + + cached.ShouldNotBeNull(); + cached.DeploymentId.ShouldBe("dep-b"); + cached.Artifact.ShouldBe(newest); + + // A re-homed node booting a neighbouring cluster's configuration must be operator-visible. + // Naming both clusters is what turns "wrong tags appeared" into a one-line diagnosis. + var warning = logger.Entries.ShouldHaveSingleItem(); + warning.Level.ShouldBe(LogLevel.Warning); + warning.Message.ShouldContain(ClusterA); + warning.Message.ShouldContain(ClusterB); + } + + public void Dispose() + { + _provider.Dispose(); + + // Real files, and pooled connections outlive the provider — clearing the pools is what + // makes the delete actually succeed. + SqliteConnection.ClearAllPools(); + + foreach (var path in new[] { _dbPath, $"{_dbPath}-wal", $"{_dbPath}-shm" }) + { + if (File.Exists(path)) + File.Delete(path); + } + } + + /// Minimal logger that records formatted messages so a test can assert on them. + private sealed class CapturingLogger : ILogger + { + public List<(LogLevel Level, string Message)> Entries { get; } = []; + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + => Entries.Add((logLevel, formatter(state, exception))); + } +}