Phase 6.1 Stream A.5/A.6 — WriteIdempotent flag on DriverAttributeInfo + Modbus/S7 tag records + FlakeyDriver integration tests
Per-tag opt-in for write-retry per docs/v2/plan.md decisions #44, #45, #143. Default is false — writes never auto-retry unless the driver author has marked the tag as safe to replay. Core.Abstractions: - DriverAttributeInfo gains `bool WriteIdempotent = false` at the end of the positional record (back-compatible; every existing call site uses the default). Driver.Modbus: - ModbusTagDefinition gains `bool WriteIdempotent = false`. Safe candidates documented in the param XML: holding-register set-points, configuration registers. Unsafe: edge-triggered coils, counter-increment addresses. - ModbusDriver.DiscoverAsync propagates t.WriteIdempotent into DriverAttributeInfo.WriteIdempotent. Driver.S7: - S7TagDefinition gains `bool WriteIdempotent = false`. Safe candidates: DB word/dword set-points, configuration DBs. Unsafe: M/Q bits that drive edge-triggered program routines. - S7Driver.DiscoverAsync propagates the flag. Stream A.5 integration tests (FlakeyDriverIntegrationTests, 4 new) exercise the invoker + flaky-driver contract the plan enumerates: - Read with 5 transient failures succeeds on the 6th attempt (RetryCount=10). - Non-idempotent write with RetryCount=5 configured still fails on the first failure — no replay (decision #44 guard at the ExecuteWriteAsync surface). - Idempotent write with 2 transient failures succeeds on the 3rd attempt. - Two hosts on the same driver have independent breakers — dead-host trips its breaker but live-host's first call still succeeds. Propagation tests: - ModbusDriverTests: SetPoint WriteIdempotent=true flows into DriverAttributeInfo; PulseCoil default=false. - S7DiscoveryAndSubscribeTests: same pattern for DBx SetPoint vs M-bit. Full solution dotnet test: 947 passing (baseline 906, +41 net across Stream A so far). Pre-existing Client.CLI Subscribe flake unchanged. Stream A's remaining work (wiring CapabilityInvoker into DriverNodeManager's OnReadValue / OnWriteValue / History / Subscribe dispatch paths) is the server-side integration piece + needs DI wiring for the pipeline builder — lands in the next PR on this branch. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,157 @@
|
||||
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;
|
||||
|
||||
/// <summary>
|
||||
/// Integration tests for the Phase 6.1 Stream A.5 contract — wrapping a flaky
|
||||
/// <see cref="IReadable"/> / <see cref="IWritable"/> through the <see cref="CapabilityInvoker"/>.
|
||||
/// Exercises the three scenarios the plan enumerates: transient read succeeds after N
|
||||
/// retries; non-idempotent write fails after one attempt; idempotent write retries through.
|
||||
/// </summary>
|
||||
[Trait("Category", "Integration")]
|
||||
public sealed class FlakeyDriverIntegrationTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Read_SurfacesSuccess_AfterTransientFailures()
|
||||
{
|
||||
var flaky = new FlakeyDriver(failReadsBeforeIndex: 5);
|
||||
var options = new DriverResilienceOptions
|
||||
{
|
||||
Tier = DriverTier.A,
|
||||
CapabilityPolicies = new Dictionary<DriverCapability, CapabilityPolicy>
|
||||
{
|
||||
[DriverCapability.Read] = new(TimeoutSeconds: 2, RetryCount: 10, BreakerFailureThreshold: 50),
|
||||
},
|
||||
};
|
||||
var invoker = new CapabilityInvoker(new DriverResiliencePipelineBuilder(), Guid.NewGuid(), () => options);
|
||||
|
||||
var result = await invoker.ExecuteAsync(
|
||||
DriverCapability.Read,
|
||||
"host-1",
|
||||
async ct => await flaky.ReadAsync(["tag-a"], ct),
|
||||
CancellationToken.None);
|
||||
|
||||
flaky.ReadAttempts.ShouldBe(6);
|
||||
result[0].StatusCode.ShouldBe(0u);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Write_NonIdempotent_FailsOnFirstFailure_NoReplay()
|
||||
{
|
||||
var flaky = new FlakeyDriver(failWritesBeforeIndex: 3);
|
||||
var optionsWithAggressiveRetry = new DriverResilienceOptions
|
||||
{
|
||||
Tier = DriverTier.A,
|
||||
CapabilityPolicies = new Dictionary<DriverCapability, CapabilityPolicy>
|
||||
{
|
||||
[DriverCapability.Write] = new(TimeoutSeconds: 2, RetryCount: 5, BreakerFailureThreshold: 50),
|
||||
},
|
||||
};
|
||||
var invoker = new CapabilityInvoker(new DriverResiliencePipelineBuilder(), Guid.NewGuid(), () => optionsWithAggressiveRetry);
|
||||
|
||||
await Should.ThrowAsync<InvalidOperationException>(async () =>
|
||||
await invoker.ExecuteWriteAsync(
|
||||
"host-1",
|
||||
isIdempotent: false,
|
||||
async ct => await flaky.WriteAsync([new WriteRequest("pulse-coil", true)], ct),
|
||||
CancellationToken.None));
|
||||
|
||||
flaky.WriteAttempts.ShouldBe(1, "non-idempotent write must never replay (decision #44)");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Write_Idempotent_RetriesUntilSuccess()
|
||||
{
|
||||
var flaky = new FlakeyDriver(failWritesBeforeIndex: 2);
|
||||
var optionsWithRetry = new DriverResilienceOptions
|
||||
{
|
||||
Tier = DriverTier.A,
|
||||
CapabilityPolicies = new Dictionary<DriverCapability, CapabilityPolicy>
|
||||
{
|
||||
[DriverCapability.Write] = new(TimeoutSeconds: 2, RetryCount: 5, BreakerFailureThreshold: 50),
|
||||
},
|
||||
};
|
||||
var invoker = new CapabilityInvoker(new DriverResiliencePipelineBuilder(), Guid.NewGuid(), () => optionsWithRetry);
|
||||
|
||||
var results = await invoker.ExecuteWriteAsync(
|
||||
"host-1",
|
||||
isIdempotent: true,
|
||||
async ct => await flaky.WriteAsync([new WriteRequest("set-point", 42.0f)], ct),
|
||||
CancellationToken.None);
|
||||
|
||||
flaky.WriteAttempts.ShouldBe(3);
|
||||
results[0].StatusCode.ShouldBe(0u);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task MultipleHosts_OnOneDriver_HaveIndependentFailureCounts()
|
||||
{
|
||||
var flaky = new FlakeyDriver(failReadsBeforeIndex: 0);
|
||||
var options = new DriverResilienceOptions { Tier = DriverTier.A };
|
||||
var builder = new DriverResiliencePipelineBuilder();
|
||||
var invoker = new CapabilityInvoker(builder, Guid.NewGuid(), () => options);
|
||||
|
||||
// host-dead: force many failures to exhaust retries + trip breaker
|
||||
var threshold = options.Resolve(DriverCapability.Read).BreakerFailureThreshold;
|
||||
for (var i = 0; i < threshold + 5; i++)
|
||||
{
|
||||
await Should.ThrowAsync<Exception>(async () =>
|
||||
await invoker.ExecuteAsync(DriverCapability.Read, "host-dead",
|
||||
_ => throw new InvalidOperationException("dead"),
|
||||
CancellationToken.None));
|
||||
}
|
||||
|
||||
// host-live: succeeds on first call — unaffected by the dead-host breaker
|
||||
var liveAttempts = 0;
|
||||
await invoker.ExecuteAsync(DriverCapability.Read, "host-live",
|
||||
_ => { liveAttempts++; return ValueTask.FromResult("ok"); },
|
||||
CancellationToken.None);
|
||||
|
||||
liveAttempts.ShouldBe(1);
|
||||
}
|
||||
|
||||
private sealed class FlakeyDriver : IReadable, IWritable
|
||||
{
|
||||
private readonly int _failReadsBeforeIndex;
|
||||
private readonly int _failWritesBeforeIndex;
|
||||
|
||||
public int ReadAttempts { get; private set; }
|
||||
public int WriteAttempts { get; private set; }
|
||||
|
||||
public FlakeyDriver(int failReadsBeforeIndex = 0, int failWritesBeforeIndex = 0)
|
||||
{
|
||||
_failReadsBeforeIndex = failReadsBeforeIndex;
|
||||
_failWritesBeforeIndex = failWritesBeforeIndex;
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
|
||||
IReadOnlyList<string> fullReferences,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var attempt = ++ReadAttempts;
|
||||
if (attempt <= _failReadsBeforeIndex)
|
||||
throw new InvalidOperationException($"transient read failure #{attempt}");
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
IReadOnlyList<DataValueSnapshot> result = fullReferences
|
||||
.Select(_ => new DataValueSnapshot(Value: 0, StatusCode: 0u, SourceTimestampUtc: now, ServerTimestampUtc: now))
|
||||
.ToList();
|
||||
return Task.FromResult(result);
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<WriteResult>> WriteAsync(
|
||||
IReadOnlyList<WriteRequest> writes,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var attempt = ++WriteAttempts;
|
||||
if (attempt <= _failWritesBeforeIndex)
|
||||
throw new InvalidOperationException($"transient write failure #{attempt}");
|
||||
|
||||
IReadOnlyList<WriteResult> result = writes.Select(_ => new WriteResult(0u)).ToList();
|
||||
return Task.FromResult(result);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -220,6 +220,23 @@ public sealed class ModbusDriverTests
|
||||
builder.Variables.ShouldContain(v => v.BrowseName == "Run" && v.Info.DriverDataType == DriverDataType.Boolean);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Discover_propagates_WriteIdempotent_from_tag_to_attribute_info()
|
||||
{
|
||||
var (drv, _) = NewDriver(
|
||||
new ModbusTagDefinition("SetPoint", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Float32, WriteIdempotent: true),
|
||||
new ModbusTagDefinition("PulseCoil", ModbusRegion.Coils, 0, ModbusDataType.Bool));
|
||||
await drv.InitializeAsync("{}", CancellationToken.None);
|
||||
|
||||
var builder = new RecordingBuilder();
|
||||
await drv.DiscoverAsync(builder, CancellationToken.None);
|
||||
|
||||
var setPoint = builder.Variables.Single(v => v.BrowseName == "SetPoint");
|
||||
var pulse = builder.Variables.Single(v => v.BrowseName == "PulseCoil");
|
||||
setPoint.Info.WriteIdempotent.ShouldBeTrue();
|
||||
pulse.Info.WriteIdempotent.ShouldBeFalse("default is opt-in per decision #44");
|
||||
}
|
||||
|
||||
// --- helpers ---
|
||||
|
||||
private sealed class RecordingBuilder : IAddressSpaceBuilder
|
||||
|
||||
@@ -65,6 +65,27 @@ public sealed class S7DiscoveryAndSubscribeTests
|
||||
builder.Variables[2].Attr.DriverDataType.ShouldBe(DriverDataType.Float32);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task DiscoverAsync_propagates_WriteIdempotent_from_tag_to_attribute_info()
|
||||
{
|
||||
var opts = new S7DriverOptions
|
||||
{
|
||||
Host = "192.0.2.1",
|
||||
Tags =
|
||||
[
|
||||
new("SetPoint", "DB1.DBW0", S7DataType.Int16, WriteIdempotent: true),
|
||||
new("StartBit", "M0.0", S7DataType.Bool),
|
||||
],
|
||||
};
|
||||
using var drv = new S7Driver(opts, "s7-idem");
|
||||
|
||||
var builder = new RecordingAddressSpaceBuilder();
|
||||
await drv.DiscoverAsync(builder, TestContext.Current.CancellationToken);
|
||||
|
||||
builder.Variables.Single(v => v.Name == "SetPoint").Attr.WriteIdempotent.ShouldBeTrue();
|
||||
builder.Variables.Single(v => v.Name == "StartBit").Attr.WriteIdempotent.ShouldBeFalse("default is opt-in per decision #44");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void GetHostStatuses_returns_one_row_with_host_port_identity_pre_init()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user