src/ZB.MOM.WW.OtOpcUa.Core/Resilience/CapabilityInvoker.cs

@@ -0,0 +1,106 @@
+using Polly;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.Resilience;
+
+/// <summary>
+///     Executes driver-capability calls through a shared Polly pipeline. One invoker per
+///     <c>(DriverInstance, IDriver)</c> pair; the underlying <see cref="DriverResiliencePipelineBuilder"/>
+///     is process-singleton so all invokers share its cache.
+/// </summary>
+/// <remarks>
+///     Per <c>docs/v2/plan.md</c> decisions #143-144 and Phase 6.1 Stream A.3. The server's dispatch
+///     layer routes every capability call (<c>IReadable.ReadAsync</c>, <c>IWritable.WriteAsync</c>,
+///     <c>ITagDiscovery.DiscoverAsync</c>, <c>ISubscribable.SubscribeAsync/UnsubscribeAsync</c>,
+///     <c>IHostConnectivityProbe</c> probe loop, <c>IAlarmSource.SubscribeAlarmsAsync/AcknowledgeAsync</c>,
+///     and all four <c>IHistoryProvider</c> reads) through this invoker.
+/// </remarks>
+public sealed class CapabilityInvoker
+{
+    private readonly DriverResiliencePipelineBuilder _builder;
+    private readonly Guid _driverInstanceId;
+    private readonly Func<DriverResilienceOptions> _optionsAccessor;
+
+    /// <summary>
+    ///     Construct an invoker for one driver instance.
+    /// </summary>
+    /// <param name="builder">Shared, process-singleton pipeline builder.</param>
+    /// <param name="driverInstanceId">The <c>DriverInstance.Id</c> column value.</param>
+    /// <param name="optionsAccessor">
+    ///     Snapshot accessor for the current resilience options. Invoked per call so Admin-edit +
+    ///     pipeline-invalidate can take effect without restarting the invoker.
+    /// </param>
+    public CapabilityInvoker(
+        DriverResiliencePipelineBuilder builder,
+        Guid driverInstanceId,
+        Func<DriverResilienceOptions> optionsAccessor)
+    {
+        ArgumentNullException.ThrowIfNull(builder);
+        ArgumentNullException.ThrowIfNull(optionsAccessor);
+
+        _builder = builder;
+        _driverInstanceId = driverInstanceId;
+        _optionsAccessor = optionsAccessor;
+    }
+
+    /// <summary>Execute a capability call returning a value, honoring the per-capability pipeline.</summary>
+    /// <typeparam name="TResult">Return type of the underlying driver call.</typeparam>
+    public async ValueTask<TResult> ExecuteAsync<TResult>(
+        DriverCapability capability,
+        string hostName,
+        Func<CancellationToken, ValueTask<TResult>> callSite,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(callSite);
+
+        var pipeline = ResolvePipeline(capability, hostName);
+        return await pipeline.ExecuteAsync(callSite, cancellationToken).ConfigureAwait(false);
+    }
+
+    /// <summary>Execute a void-returning capability call, honoring the per-capability pipeline.</summary>
+    public async ValueTask ExecuteAsync(
+        DriverCapability capability,
+        string hostName,
+        Func<CancellationToken, ValueTask> callSite,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(callSite);
+
+        var pipeline = ResolvePipeline(capability, hostName);
+        await pipeline.ExecuteAsync(callSite, cancellationToken).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    ///     Execute a <see cref="DriverCapability.Write"/> call honoring <see cref="WriteIdempotentAttribute"/>
+    ///     semantics — if <paramref name="isIdempotent"/> is <c>false</c>, retries are disabled regardless
+    ///     of the tag-level configuration (the pipeline for a non-idempotent write never retries per
+    ///     decisions #44-45). If <c>true</c>, the call runs through the capability's pipeline which may
+    ///     retry when the tier configuration permits.
+    /// </summary>
+    public async ValueTask<TResult> ExecuteWriteAsync<TResult>(
+        string hostName,
+        bool isIdempotent,
+        Func<CancellationToken, ValueTask<TResult>> callSite,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(callSite);
+
+        if (!isIdempotent)
+        {
+            var noRetryOptions = _optionsAccessor() with
+            {
+                CapabilityPolicies = new Dictionary<DriverCapability, CapabilityPolicy>
+                {
+                    [DriverCapability.Write] = _optionsAccessor().Resolve(DriverCapability.Write) with { RetryCount = 0 },
+                },
+            };
+            var pipeline = _builder.GetOrCreate(_driverInstanceId, $"{hostName}::non-idempotent", DriverCapability.Write, noRetryOptions);
+            return await pipeline.ExecuteAsync(callSite, cancellationToken).ConfigureAwait(false);
+        }
+
+        return await ExecuteAsync(DriverCapability.Write, hostName, callSite, cancellationToken).ConfigureAwait(false);
+    }
+
+    private ResiliencePipeline ResolvePipeline(DriverCapability capability, string hostName) =>
+        _builder.GetOrCreate(_driverInstanceId, hostName, capability, _optionsAccessor());
+}

tests/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/CapabilityInvokerTests.cs

@@ -0,0 +1,151 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.Resilience;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.Tests.Resilience;
+
+[Trait("Category", "Unit")]
+public sealed class CapabilityInvokerTests
+{
+    private static CapabilityInvoker MakeInvoker(
+        DriverResiliencePipelineBuilder builder,
+        DriverResilienceOptions options) =>
+        new(builder, Guid.NewGuid(), () => options);
+
+    [Fact]
+    public async Task Read_ReturnsValue_FromCallSite()
+    {
+        var invoker = MakeInvoker(new DriverResiliencePipelineBuilder(), new DriverResilienceOptions { Tier = DriverTier.A });
+
+        var result = await invoker.ExecuteAsync(
+            DriverCapability.Read,
+            "host-1",
+            _ => ValueTask.FromResult(42),
+            CancellationToken.None);
+
+        result.ShouldBe(42);
+    }
+
+    [Fact]
+    public async Task Read_Retries_OnTransientFailure()
+    {
+        var invoker = MakeInvoker(new DriverResiliencePipelineBuilder(), new DriverResilienceOptions { Tier = DriverTier.A });
+        var attempts = 0;
+
+        var result = await invoker.ExecuteAsync(
+            DriverCapability.Read,
+            "host-1",
+            async _ =>
+            {
+                attempts++;
+                if (attempts < 2) throw new InvalidOperationException("transient");
+                await Task.Yield();
+                return "ok";
+            },
+            CancellationToken.None);
+
+        result.ShouldBe("ok");
+        attempts.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task Write_NonIdempotent_DoesNotRetry_EvenWhenPolicyHasRetries()
+    {
+        var options = new DriverResilienceOptions
+        {
+            Tier = DriverTier.A,
+            CapabilityPolicies = new Dictionary<DriverCapability, CapabilityPolicy>
+            {
+                [DriverCapability.Write] = new(TimeoutSeconds: 2, RetryCount: 3, BreakerFailureThreshold: 5),
+            },
+        };
+        var invoker = MakeInvoker(new DriverResiliencePipelineBuilder(), options);
+        var attempts = 0;
+
+        await Should.ThrowAsync<InvalidOperationException>(async () =>
+            await invoker.ExecuteWriteAsync(
+                "host-1",
+                isIdempotent: false,
+                async _ =>
+                {
+                    attempts++;
+                    await Task.Yield();
+                    throw new InvalidOperationException("boom");
+#pragma warning disable CS0162
+                    return 0;
+#pragma warning restore CS0162
+                },
+                CancellationToken.None));
+
+        attempts.ShouldBe(1, "non-idempotent write must never replay");
+    }
+
+    [Fact]
+    public async Task Write_Idempotent_Retries_WhenPolicyHasRetries()
+    {
+        var options = new DriverResilienceOptions
+        {
+            Tier = DriverTier.A,
+            CapabilityPolicies = new Dictionary<DriverCapability, CapabilityPolicy>
+            {
+                [DriverCapability.Write] = new(TimeoutSeconds: 2, RetryCount: 3, BreakerFailureThreshold: 5),
+            },
+        };
+        var invoker = MakeInvoker(new DriverResiliencePipelineBuilder(), options);
+        var attempts = 0;
+
+        var result = await invoker.ExecuteWriteAsync(
+            "host-1",
+            isIdempotent: true,
+            async _ =>
+            {
+                attempts++;
+                if (attempts < 2) throw new InvalidOperationException("transient");
+                await Task.Yield();
+                return "ok";
+            },
+            CancellationToken.None);
+
+        result.ShouldBe("ok");
+        attempts.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task Write_Default_DoesNotRetry_WhenPolicyHasZeroRetries()
+    {
+        // Tier A Write default is RetryCount=0. Even isIdempotent=true shouldn't retry
+        // because the policy says not to.
+        var invoker = MakeInvoker(new DriverResiliencePipelineBuilder(), new DriverResilienceOptions { Tier = DriverTier.A });
+        var attempts = 0;
+
+        await Should.ThrowAsync<InvalidOperationException>(async () =>
+            await invoker.ExecuteWriteAsync(
+                "host-1",
+                isIdempotent: true,
+                async _ =>
+                {
+                    attempts++;
+                    await Task.Yield();
+                    throw new InvalidOperationException("boom");
+#pragma warning disable CS0162
+                    return 0;
+#pragma warning restore CS0162
+                },
+                CancellationToken.None));
+
+        attempts.ShouldBe(1, "tier-A default for Write is RetryCount=0");
+    }
+
+    [Fact]
+    public async Task Execute_HonorsDifferentHosts_Independently()
+    {
+        var builder = new DriverResiliencePipelineBuilder();
+        var invoker = MakeInvoker(builder, new DriverResilienceOptions { Tier = DriverTier.A });
+
+        await invoker.ExecuteAsync(DriverCapability.Read, "host-a", _ => ValueTask.FromResult(1), CancellationToken.None);
+        await invoker.ExecuteAsync(DriverCapability.Read, "host-b", _ => ValueTask.FromResult(2), CancellationToken.None);
+
+        builder.CachedPipelineCount.ShouldBe(2);
+    }
+}