feat(localdb): dedicated h2c listener + MapZbLocalDbSync gated on LocalDb:SyncListenPort
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,68 @@
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Deployment;
|
||||
|
||||
/// <summary>
|
||||
/// Node-local durable cache of the deployment artifact this node last applied.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public interface IDeploymentArtifactCache
|
||||
{
|
||||
/// <summary>
|
||||
/// Persists <paramref name="artifact"/> as the cluster's current deployment.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Idempotent per <paramref name="deploymentId"/>: 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
|
||||
byte[] artifact, CancellationToken ct = default);
|
||||
|
||||
/// <summary>
|
||||
/// Reads the cached artifact for <paramref name="clusterId"/>, or <see langword="null"/>
|
||||
/// when nothing is cached or the cached bytes fail their integrity check.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
Task<CachedDeploymentArtifact?> GetCurrentAsync(string clusterId, CancellationToken ct = default);
|
||||
|
||||
/// <summary>
|
||||
/// Reads the single cached pointer without knowing the cluster id.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public sealed record CachedDeploymentArtifact(
|
||||
string DeploymentId, string RevisionHash, byte[] Artifact, DateTimeOffset AppliedAtUtc);
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="IDeploymentArtifactCache"/> backed by the replicated LocalDb deployment-cache
|
||||
/// tables.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Why the artifact is stored as base64 chunks.</b> See
|
||||
/// <see cref="DeploymentCacheSchema"/> — 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class LocalDbDeploymentArtifactCache : IDeploymentArtifactCache
|
||||
{
|
||||
/// <summary>
|
||||
/// Raw bytes per chunk, before base64 expansion.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
private const int ChunkSize = 128 * 1024;
|
||||
|
||||
/// <summary>
|
||||
/// How many deployments per cluster survive a store.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
private const int RetainedDeployments = 2;
|
||||
|
||||
private readonly ILocalDb _db;
|
||||
private readonly ILogger<LocalDbDeploymentArtifactCache> _logger;
|
||||
|
||||
public LocalDbDeploymentArtifactCache(ILocalDb db, ILogger<LocalDbDeploymentArtifactCache> logger)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(db);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
|
||||
_db = db;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task<CachedDeploymentArtifact?> 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);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task<CachedDeploymentArtifact?> 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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Rebuilds the artifact the pointer names, returning <see langword="null"/> unless every
|
||||
/// integrity check passes.
|
||||
/// </summary>
|
||||
private async Task<CachedDeploymentArtifact?> 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<string> 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));
|
||||
|
||||
/// <summary>
|
||||
/// Round-trip ("O") UTC, so lexicographic ordering of the stored TEXT is chronological
|
||||
/// ordering — the retention query sorts on it directly.
|
||||
/// </summary>
|
||||
private static string FormatTimestamp(DateTimeOffset value)
|
||||
=> value.ToUniversalTime().ToString("O", CultureInfo.InvariantCulture);
|
||||
|
||||
private sealed record PointerRow(
|
||||
string DeploymentId, string RevisionHash, string ArtifactSha256, string AppliedAtUtc);
|
||||
}
|
||||
Reference in New Issue
Block a user