9cad9ed0fc
v2-ci / build (push) Failing after 41s
v2-ci / unit-tests (tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Security.Tests) (push) Has been skipped
v2-ci / integration (tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests) (push) Has been skipped
v2-ci / integration (tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests) (push) Has been skipped
Adds <summary>/<param>/<returns>/<inheritdoc> where missing and removes project bookkeeping IDs (task/tracking refs) from shipped code comments, so the docs read cleanly and CommentChecker is quiet except for known false positives (PLC/protocol terms, event/IEqualityComparer inheritdoc). Doc/comment-only; no logic changed; solution builds clean.
141 lines
7.0 KiB
C#
141 lines
7.0 KiB
C#
using System.Text.RegularExpressions;
|
||
using Microsoft.Extensions.Logging;
|
||
using Polly;
|
||
using Polly.Retry;
|
||
using Polly.Timeout;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Configuration.LocalCache;
|
||
|
||
/// <summary>
|
||
/// Wraps a central-DB fetch function with Phase 6.1 Stream D.2 resilience:
|
||
/// <b>timeout 2 s → retry 3× jittered → fallback to sealed cache</b>. Maintains the
|
||
/// <see cref="StaleConfigFlag"/> — fresh on central-DB success, stale on cache fallback.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>Read-path only per plan. The write path (draft save, publish) bypasses this
|
||
/// wrapper entirely and fails hard on DB outage so inconsistent writes never land.</para>
|
||
///
|
||
/// <para>Fallback is triggered by <b>any exception</b> the fetch raises (central-DB
|
||
/// unreachable, SqlException, timeout). If the sealed cache also fails (no pointer,
|
||
/// corrupt file, etc.), <see cref="GenerationCacheUnavailableException"/> surfaces — caller
|
||
/// must fail the current request (InitializeAsync for a driver, etc.).</para>
|
||
/// </remarks>
|
||
public sealed class ResilientConfigReader
|
||
{
|
||
private readonly GenerationSealedCache _cache;
|
||
private readonly StaleConfigFlag _staleFlag;
|
||
private readonly ResiliencePipeline _pipeline;
|
||
private readonly ILogger<ResilientConfigReader> _logger;
|
||
|
||
/// <summary>Initializes a resilient config reader with the given cache and options.</summary>
|
||
/// <param name="cache">The sealed cache for fallback.</param>
|
||
/// <param name="staleFlag">The stale config flag to manage.</param>
|
||
/// <param name="logger">The logger instance.</param>
|
||
/// <param name="timeout">The timeout for central fetch (default 2s).</param>
|
||
/// <param name="retryCount">The number of retries (default 3).</param>
|
||
public ResilientConfigReader(
|
||
GenerationSealedCache cache,
|
||
StaleConfigFlag staleFlag,
|
||
ILogger<ResilientConfigReader> logger,
|
||
TimeSpan? timeout = null,
|
||
int retryCount = 3)
|
||
{
|
||
_cache = cache;
|
||
_staleFlag = staleFlag;
|
||
_logger = logger;
|
||
var builder = new ResiliencePipelineBuilder()
|
||
.AddTimeout(new TimeoutStrategyOptions { Timeout = timeout ?? TimeSpan.FromSeconds(2) });
|
||
|
||
if (retryCount > 0)
|
||
{
|
||
builder.AddRetry(new RetryStrategyOptions
|
||
{
|
||
MaxRetryAttempts = retryCount,
|
||
BackoffType = DelayBackoffType.Exponential,
|
||
UseJitter = true,
|
||
Delay = TimeSpan.FromMilliseconds(100),
|
||
MaxDelay = TimeSpan.FromSeconds(1),
|
||
// Handle ALL exceptions including OperationCanceledException. A SQL command-level
|
||
// timeout surfaces as TaskCanceledException (derives from OperationCanceledException)
|
||
// when the caller's token is NOT cancelled, and must be retried just like any other
|
||
// transient error. Polly itself checks the cancellation token between retries and
|
||
// stops with OperationCanceledException on genuine caller cancellation regardless of
|
||
// this predicate.
|
||
ShouldHandle = new PredicateBuilder().Handle<Exception>(),
|
||
});
|
||
}
|
||
|
||
_pipeline = builder.Build();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Redacts connection-string fragments (Password, User Id, Pwd, etc.)
|
||
/// that a caller's exception message could carry. Conservative regex pass — anything
|
||
/// matching <c>Key=Value</c> with a known credential key gets its value replaced.
|
||
/// </summary>
|
||
private static readonly Regex SecretsRegex = new(
|
||
@"(?ix)\b(Password|Pwd|User\s*Id|Uid|AccessToken|Authorization|Api[-_]?Key)\s*=\s*[^;,)\s]*",
|
||
RegexOptions.Compiled);
|
||
|
||
/// <summary>Redacts sensitive credential information from a message.</summary>
|
||
/// <param name="message">The message to scrub.</param>
|
||
/// <returns>The message with redacted credentials.</returns>
|
||
internal static string ScrubSecrets(string? message)
|
||
{
|
||
if (string.IsNullOrEmpty(message)) return message ?? string.Empty;
|
||
// Replace the entire matched fragment (key + value) with a redaction marker so the
|
||
// key name itself doesn't leak — log scrapers grep for "Password=" too.
|
||
return SecretsRegex.Replace(message, "[redacted credential]");
|
||
}
|
||
|
||
/// <summary>
|
||
/// Executes a central fetch through the resilience pipeline. On full failure
|
||
/// (post-retry), reads the sealed cache and extracts the requested shape.
|
||
/// </summary>
|
||
/// <typeparam name="T">The type of configuration to read.</typeparam>
|
||
/// <param name="clusterId">The cluster ID to fetch for.</param>
|
||
/// <param name="centralFetch">Function to fetch from central DB.</param>
|
||
/// <param name="fromSnapshot">Function to extract the config from a snapshot.</param>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>The configuration of type T.</returns>
|
||
public async ValueTask<T> ReadAsync<T>(
|
||
string clusterId,
|
||
Func<CancellationToken, ValueTask<T>> centralFetch,
|
||
Func<GenerationSnapshot, T> fromSnapshot,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
ArgumentException.ThrowIfNullOrWhiteSpace(clusterId);
|
||
ArgumentNullException.ThrowIfNull(centralFetch);
|
||
ArgumentNullException.ThrowIfNull(fromSnapshot);
|
||
|
||
try
|
||
{
|
||
var result = await _pipeline.ExecuteAsync(centralFetch, cancellationToken).ConfigureAwait(false);
|
||
_staleFlag.MarkFresh();
|
||
return result;
|
||
}
|
||
// Catch all exceptions that are NOT genuine caller cancellations. A SQL command-level
|
||
// timeout surfaces as TaskCanceledException (derives from OperationCanceledException)
|
||
// but the caller's token is NOT cancelled — we must fall back to the sealed cache for
|
||
// that case, not propagate. Only rethrow if the caller actually requested cancellation.
|
||
catch (Exception ex) when (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequested)
|
||
{
|
||
// Do NOT pass the raw exception object — it carries the stack
|
||
// and inner-exception chain, and SqlException/wrapping delegates can surface
|
||
// connection-string fragments (Password=…, User Id=…) embedded in messages.
|
||
// Log only the exception type and a scrubbed message so secrets stay out of logs.
|
||
_logger.LogWarning(
|
||
"Central-DB read failed after retries ({ExceptionType}: {SanitizedMessage}); falling back to sealed cache for cluster {ClusterId}",
|
||
ex.GetType().Name,
|
||
ScrubSecrets(ex.Message),
|
||
clusterId);
|
||
// GenerationCacheUnavailableException surfaces intentionally — fails the caller's
|
||
// operation. StaleConfigFlag stays unchanged; the flag only flips when we actually
|
||
// served a cache snapshot.
|
||
var snapshot = await _cache.ReadCurrentAsync(clusterId, cancellationToken).ConfigureAwait(false);
|
||
_staleFlag.MarkStale();
|
||
return fromSnapshot(snapshot);
|
||
}
|
||
}
|
||
}
|