using System.Collections.Concurrent;
using Microsoft.Extensions.Logging;
using Polly;
using Polly.CircuitBreaker;
using Polly.Retry;
using Polly.Timeout;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.Resilience;
///
/// Builds and caches Polly resilience pipelines keyed on
/// (DriverInstanceId, HostName, DriverCapability). One dead PLC behind a multi-device
/// driver cannot open the circuit breaker for healthy sibling hosts.
///
///
/// Per-device isolation. Composition from outside-in:
/// Timeout → Retry (when capability permits) → Circuit Breaker (when tier permits).
///
/// Pipeline resolution is lock-free on the hot path: the inner
/// caches a per key;
/// first-call cost is one .Build. Thereafter reads are O(1).
///
/// Never-throws invariant (01/S-6): clamps every policy value to
/// before constructing strategy options, so a hostile
/// (constructed directly, bypassing the parser) can never make
/// the memoizing GetOrAdd factory throw Polly's ValidationException — which would
/// otherwise brick every capability call for that key.
///
public sealed class DriverResiliencePipelineBuilder
{
private readonly ConcurrentDictionary _pipelines = new();
private readonly TimeProvider _timeProvider;
private readonly DriverResilienceStatusTracker? _statusTracker;
private readonly ILogger? _logger;
/// Construct with the ambient clock (use in prod).
/// Clock source for pipeline timeouts + breaker sampling. Defaults to system.
/// When non-null, every built pipeline wires Polly telemetry into
/// the tracker — retries increment ConsecutiveFailures, breaker-open stamps
/// LastBreakerOpenUtc, breaker-close resets failures. Feeds Admin /hosts +
/// the in-flight-call-depth column. Absent tracker means no telemetry (unit tests +
/// deployments that don't care about resilience observability).
/// When non-null, retry / circuit-breaker-open / breaker-close events are
/// logged (instance + host + capability + attempt/exception). This is the operator-facing
/// observability surface for the resilience pipeline until the Admin /hosts reader
/// (Phase 6.1 Stream E.2/E.3) ships — without it the pipeline runs silently. Absent logger =
/// no resilience logging (unit tests).
public DriverResiliencePipelineBuilder(
TimeProvider? timeProvider = null,
DriverResilienceStatusTracker? statusTracker = null,
ILogger? logger = null)
{
_timeProvider = timeProvider ?? TimeProvider.System;
_statusTracker = statusTracker;
_logger = logger;
}
///
/// Get or build the pipeline for a given (driver instance, host, capability) triple.
/// Calls with the same key + same options reuse the same pipeline instance; the first caller
/// wins if a race occurs (both pipelines would be behaviourally identical).
///
/// DriverInstance primary key — opaque to this layer.
///
/// Host the call targets. For single-host drivers (Galaxy, some OPC UA Client configs) pass the
/// driver's canonical host string. For multi-host drivers (Modbus with N PLCs), pass the
/// specific PLC so one dead PLC doesn't poison healthy siblings.
///
/// Which capability surface is being called.
/// Per-driver-instance options (tier + per-capability overrides).
///
/// Monotonic options epoch stamped by per invoker
/// (01/S-7). Part of the cache key so a lingering old child's post-invalidate re-cache lands under
/// its OWN (old) generation — a key the new invoker's generation never reads. Defaults to 0 so
/// every existing caller/test compiles unchanged.
///
/// The cached or newly built resilience pipeline for the key.
public ResiliencePipeline GetOrCreate(
string driverInstanceId,
string hostName,
DriverCapability capability,
DriverResilienceOptions options,
long optionsGeneration = 0)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentException.ThrowIfNullOrWhiteSpace(hostName);
var key = new PipelineKey(driverInstanceId, hostName, capability, optionsGeneration);
return _pipelines.GetOrAdd(key, static (k, state) => Build(
k.DriverInstanceId, k.HostName, state.capability, state.options, state.timeProvider, state.tracker, state.logger),
(capability, options, timeProvider: _timeProvider, tracker: _statusTracker, logger: _logger));
}
///
/// Drop cached pipelines for one driver instance (e.g. on ResilienceConfig change), across ALL
/// option generations. With generation-keyed entries (01/S-7) correctness no longer depends on this
/// sweep — a stale-epoch re-cache is already unreachable by the new generation — but it bounds the
/// cache by clearing lingering old-generation entries on the next respawn. Test + Admin-reload use.
///
/// The driver instance ID whose pipelines should be invalidated.
/// The number of pipelines removed.
public int Invalidate(string driverInstanceId)
{
var removed = 0;
foreach (var key in _pipelines.Keys)
{
if (key.DriverInstanceId == driverInstanceId && _pipelines.TryRemove(key, out _))
removed++;
}
return removed;
}
/// Snapshot of the current number of cached pipelines. For diagnostics only.
public int CachedPipelineCount => _pipelines.Count;
private static ResiliencePipeline Build(
string driverInstanceId,
string hostName,
DriverCapability capability,
DriverResilienceOptions options,
TimeProvider timeProvider,
DriverResilienceStatusTracker? tracker,
ILogger? logger)
{
var policy = options.Resolve(capability);
var builder = new ResiliencePipelineBuilder { TimeProvider = timeProvider };
// Belt-and-braces: clamp the resolved policy to ResiliencePolicyRanges so building a pipeline can
// NEVER throw Polly's ValidationException (01/S-6). The parser already clamps every JSON→options
// funnel, but DriverResilienceOptions is a public record constructible by paths that never traverse
// the parser (tests today; any future caller), so this makes "a pipeline build never throws" a
// builder invariant. Three integer ops once per pipeline build (not per call).
var timeoutSeconds = Math.Clamp(
policy.TimeoutSeconds, ResiliencePolicyRanges.MinTimeoutSeconds, ResiliencePolicyRanges.MaxTimeoutSeconds);
builder.AddTimeout(new TimeoutStrategyOptions
{
Timeout = TimeSpan.FromSeconds(timeoutSeconds),
});
if (policy.RetryCount > 0)
{
var retryOptions = new RetryStrategyOptions
{
MaxRetryAttempts = Math.Min(policy.RetryCount, ResiliencePolicyRanges.MaxRetryCount),
BackoffType = DelayBackoffType.Exponential,
UseJitter = true,
Delay = TimeSpan.FromMilliseconds(100),
MaxDelay = TimeSpan.FromSeconds(5),
ShouldHandle = new PredicateBuilder().Handle(ex => ex is not OperationCanceledException),
};
if (tracker is not null || logger is not null)
{
retryOptions.OnRetry = args =>
{
tracker?.RecordFailure(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime);
logger?.LogWarning(
"Driver resilience retry: instance={DriverInstanceId} host={HostName} capability={Capability} attempt={Attempt}: {Reason}",
driverInstanceId, hostName, capability, args.AttemptNumber, args.Outcome.Exception?.Message ?? "unknown");
return default;
};
}
builder.AddRetry(retryOptions);
}
if (policy.BreakerFailureThreshold > 0)
{
var breakerOptions = new CircuitBreakerStrategyOptions
{
FailureRatio = 1.0,
MinimumThroughput = Math.Max(ResiliencePolicyRanges.MinEnabledBreakerFailureThreshold, policy.BreakerFailureThreshold),
SamplingDuration = TimeSpan.FromSeconds(30),
BreakDuration = TimeSpan.FromSeconds(15),
ShouldHandle = new PredicateBuilder().Handle(ex => ex is not OperationCanceledException),
};
if (tracker is not null || logger is not null)
{
breakerOptions.OnOpened = args =>
{
tracker?.RecordBreakerOpen(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime);
logger?.LogWarning(
"Driver resilience circuit-breaker OPENED: instance={DriverInstanceId} host={HostName} capability={Capability} breakDuration={BreakDuration}: {Reason}",
driverInstanceId, hostName, capability, args.BreakDuration, args.Outcome.Exception?.Message ?? "unknown");
return default;
};
breakerOptions.OnClosed = args =>
{
// Closing the breaker means the target recovered — reset the consecutive-
// failure counter so Admin UI stops flashing red for this host.
tracker?.RecordSuccess(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime);
logger?.LogInformation(
"Driver resilience circuit-breaker closed (recovered): instance={DriverInstanceId} host={HostName} capability={Capability}",
driverInstanceId, hostName, capability);
return default;
};
}
builder.AddCircuitBreaker(breakerOptions);
}
return builder.Build();
}
private readonly record struct PipelineKey(string DriverInstanceId, string HostName, DriverCapability Capability, long Generation);
}