feat(archreview #13): apply per-instance ResilienceConfig from the deploy artifact
The DriverInstance.ResilienceConfig column was authored in AdminUI, persisted to the entity, and serialized into the deployment artifact by ConfigComposer — but the runtime read path dropped it: DriverInstanceSpec didn't carry it and the invoker factory always passed null, so every driver got tier defaults regardless of its configured overrides (a silent dead-config gap — #10's residual sub-finding). Read-path plumbing (write side was already complete): - DriverInstanceSpec gains ResilienceConfig; DeploymentArtifact.TryReadSpec reads the column. - IDriverCapabilityInvokerFactory.Create takes resilienceConfigJson; DriverHostActor.SpawnChild threads spec.ResilienceConfig; the concrete factory parses it (ParseOrDefaults, layering on the tier), logs any parse diagnostic (never throws), and builds the invoker with the merged options. - Invalidate-on-change: the pipeline cache keys on (instance, host, capability) and ignores options on a hit, so Create() now Invalidate()s the instance's cached pipelines first (no-op on first spawn) — a respawn with changed options rebuilds them. - DriverSpawnPlanner treats a ResilienceConfig change as a stop+respawn (the invoker/options are bound to the child at spawn); a pure DriverConfig change stays an in-place delta (no reconnect). - Host DI passes a logger to the factory for the parse diagnostic. Verification (deterministic): factory Create applies a retryCount:0 override to actual execution (control test proves tier default retries), invalidates the instance's cache on re-create (scoped — sibling survives), malformed config logs+falls-back; planner respawns on ResilienceConfig change (incl null→json) and stays delta on a pure config change; artifact parse carries/omits the column. Core.Tests 243 (+5), Runtime.Tests 363 (+5), Host builds clean.
This commit is contained in:
Joseph Doherty
+148
@@ -0,0 +1,148 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
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>
|
||||
/// Guards the per-instance ResilienceConfig plumbing (arch-review #13): the factory must layer the
|
||||
/// per-instance <c>ResilienceConfig</c> JSON onto the tier defaults, invalidate any stale cached
|
||||
/// pipelines for the instance on every <see cref="DriverCapabilityInvokerFactory.Create"/> (so a
|
||||
/// respawn with a changed config takes effect despite the options-blind pipeline cache), and log —
|
||||
/// never throw on — a malformed config.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class DriverCapabilityInvokerFactoryTests
|
||||
{
|
||||
private static DriverCapabilityInvokerFactory MakeFactory(
|
||||
DriverResiliencePipelineBuilder builder,
|
||||
DriverTier tier = DriverTier.A,
|
||||
ILogger? logger = null) =>
|
||||
new(builder, new DriverResilienceStatusTracker(), _ => tier, logger);
|
||||
|
||||
/// <summary>
|
||||
/// End-to-end proof the per-instance override REACHES execution: a ResilienceConfig that sets
|
||||
/// Read's retryCount to 0 must suppress the tier-A Read retry — a failing Read is attempted once,
|
||||
/// not retried.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Create_applies_per_instance_ResilienceConfig_override_to_behaviour()
|
||||
{
|
||||
var builder = new DriverResiliencePipelineBuilder();
|
||||
var factory = MakeFactory(builder);
|
||||
// Tier A Read retries by default (see CapabilityInvokerTests.Read_Retries_OnTransientFailure);
|
||||
// this override disables it.
|
||||
var invoker = factory.Create("d1", "Modbus",
|
||||
resilienceConfigJson: "{\"capabilityPolicies\":{\"Read\":{\"retryCount\":0}}}");
|
||||
var attempts = 0;
|
||||
|
||||
await Should.ThrowAsync<InvalidOperationException>(async () =>
|
||||
await invoker.ExecuteAsync(
|
||||
DriverCapability.Read,
|
||||
"host-1",
|
||||
async _ =>
|
||||
{
|
||||
attempts++;
|
||||
await Task.Yield();
|
||||
throw new InvalidOperationException("boom");
|
||||
},
|
||||
CancellationToken.None));
|
||||
|
||||
attempts.ShouldBe(1, "the per-instance ResilienceConfig override (retryCount:0) must suppress the tier-A Read retry");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A control against the override test: with NO ResilienceConfig the same failing Read retries per
|
||||
/// the tier-A default — confirming the previous test's single attempt is the override, not a fluke.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Create_without_ResilienceConfig_uses_tier_default_retry()
|
||||
{
|
||||
var factory = MakeFactory(new DriverResiliencePipelineBuilder());
|
||||
var invoker = factory.Create("d1", "Modbus");
|
||||
var attempts = 0;
|
||||
|
||||
await Should.ThrowAsync<InvalidOperationException>(async () =>
|
||||
await invoker.ExecuteAsync(
|
||||
DriverCapability.Read,
|
||||
"host-1",
|
||||
async _ =>
|
||||
{
|
||||
attempts++;
|
||||
await Task.Yield();
|
||||
throw new InvalidOperationException("boom");
|
||||
},
|
||||
CancellationToken.None));
|
||||
|
||||
attempts.ShouldBeGreaterThan(1, "tier-A Read retries by default when no override disables it");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Invalidate-on-change: because the pipeline cache is keyed on (instance, host, capability) and
|
||||
/// ignores options on a hit, a fresh <see cref="DriverCapabilityInvokerFactory.Create"/> for an
|
||||
/// instance must drop that instance's cached pipelines so a respawn with new options rebuilds them.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Create_invalidates_prior_cached_pipelines_for_the_same_instance()
|
||||
{
|
||||
var builder = new DriverResiliencePipelineBuilder();
|
||||
var factory = MakeFactory(builder);
|
||||
|
||||
// First invoker warms the cache for instance d1.
|
||||
var first = factory.Create("d1", "Modbus");
|
||||
await first.ExecuteAsync(DriverCapability.Read, "host-1", _ => ValueTask.FromResult(1), CancellationToken.None);
|
||||
builder.CachedPipelineCount.ShouldBe(1);
|
||||
|
||||
// Re-creating the invoker for the SAME instance (a respawn) must invalidate the stale pipeline.
|
||||
_ = factory.Create("d1", "Modbus", resilienceConfigJson: "{\"bulkheadMaxConcurrent\":8}");
|
||||
|
||||
builder.CachedPipelineCount.ShouldBe(0, "Create must invalidate the instance's cached pipelines so changed options take effect");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Invalidation is scoped to the instance being (re)created — a sibling instance's warm pipeline
|
||||
/// must survive.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Create_does_not_invalidate_a_different_instance()
|
||||
{
|
||||
var builder = new DriverResiliencePipelineBuilder();
|
||||
var factory = MakeFactory(builder);
|
||||
|
||||
var other = factory.Create("other", "Modbus");
|
||||
await other.ExecuteAsync(DriverCapability.Read, "host-1", _ => ValueTask.FromResult(1), CancellationToken.None);
|
||||
builder.CachedPipelineCount.ShouldBe(1);
|
||||
|
||||
_ = factory.Create("d1", "Modbus");
|
||||
|
||||
builder.CachedPipelineCount.ShouldBe(1, "invalidation is per-instance — the sibling's pipeline must remain");
|
||||
}
|
||||
|
||||
/// <summary>A malformed ResilienceConfig logs a diagnostic but still yields a working (tier-default) invoker.</summary>
|
||||
[Fact]
|
||||
public async Task Create_with_malformed_ResilienceConfig_logs_and_falls_back_to_tier_defaults()
|
||||
{
|
||||
var logger = new CapturingLogger();
|
||||
var factory = MakeFactory(new DriverResiliencePipelineBuilder(), logger: logger);
|
||||
|
||||
var invoker = factory.Create("d1", "Modbus", resilienceConfigJson: "{ this is not json ]");
|
||||
|
||||
invoker.ShouldNotBeNull();
|
||||
// Still functional: a normal Read succeeds under the fallback tier defaults.
|
||||
var result = await invoker.ExecuteAsync(DriverCapability.Read, "host-1", _ => ValueTask.FromResult(7), CancellationToken.None);
|
||||
result.ShouldBe(7);
|
||||
logger.Entries.ShouldContain(e => e.Level == LogLevel.Warning && e.Message.Contains("resilience config", StringComparison.OrdinalIgnoreCase));
|
||||
}
|
||||
|
||||
private sealed class CapturingLogger : ILogger
|
||||
{
|
||||
public List<(LogLevel Level, string Message)> Entries { get; } = new();
|
||||
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||
public bool IsEnabled(LogLevel logLevel) => true;
|
||||
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
|
||||
Func<TState, Exception?, string> formatter) =>
|
||||
Entries.Add((logLevel, formatter(state, exception)));
|
||||
}
|
||||
}
|
||||
@@ -650,4 +650,39 @@ public sealed class DeploymentArtifactTests
|
||||
var comp = DeploymentArtifact.ParseComposition(blob, "c1-1:4053");
|
||||
comp.EquipmentNodes.Select(e => e.EquipmentId).ShouldBe(new[] { "E-c1" });
|
||||
}
|
||||
|
||||
/// <summary>Verifies the per-instance ResilienceConfig column is carried onto the parsed spec.</summary>
|
||||
[Fact]
|
||||
public void ParseDriverInstances_carries_ResilienceConfig_onto_the_spec()
|
||||
{
|
||||
var resilienceJson = "{\"bulkheadMaxConcurrent\":16,\"capabilityPolicies\":{\"Write\":{\"retryCount\":0}}}";
|
||||
var blob = BlobOf(new
|
||||
{
|
||||
DriverInstances = new[]
|
||||
{
|
||||
new { DriverInstanceId = "d1", DriverType = "Modbus", Enabled = true, DriverConfig = "{}", ResilienceConfig = resilienceJson },
|
||||
},
|
||||
});
|
||||
|
||||
var spec = DeploymentArtifact.ParseDriverInstances(blob).Single();
|
||||
|
||||
spec.ResilienceConfig.ShouldBe(resilienceJson);
|
||||
}
|
||||
|
||||
/// <summary>Verifies an absent ResilienceConfig column leaves the spec field null (⇒ tier defaults at spawn).</summary>
|
||||
[Fact]
|
||||
public void ParseDriverInstances_absent_ResilienceConfig_is_null()
|
||||
{
|
||||
var blob = BlobOf(new
|
||||
{
|
||||
DriverInstances = new[]
|
||||
{
|
||||
new { DriverInstanceId = "d1", DriverType = "Modbus", Enabled = true, DriverConfig = "{}" },
|
||||
},
|
||||
});
|
||||
|
||||
var spec = DeploymentArtifact.ParseDriverInstances(blob).Single();
|
||||
|
||||
spec.ResilienceConfig.ShouldBeNull();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,8 +6,8 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
public sealed class DriverSpawnPlannerTests
|
||||
{
|
||||
private static DriverInstanceSpec Spec(string id, string type = "Modbus", string config = "{\"host\":\"127.0.0.1\"}", bool enabled = true) =>
|
||||
new(Guid.NewGuid(), id, id, type, enabled, config);
|
||||
private static DriverInstanceSpec Spec(string id, string type = "Modbus", string config = "{\"host\":\"127.0.0.1\"}", bool enabled = true, string? resilience = null) =>
|
||||
new(Guid.NewGuid(), id, id, type, enabled, config, ClusterId: null, ResilienceConfig: resilience);
|
||||
|
||||
/// <summary>Verifies that all new drivers are placed in ToSpawn when current is empty.</summary>
|
||||
[Fact]
|
||||
@@ -122,4 +122,66 @@ public sealed class DriverSpawnPlannerTests
|
||||
plan.ToSpawn.Single().DriverType.ShouldBe("AbCip");
|
||||
plan.ToApplyDelta.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A ResilienceConfig change forces a stop+respawn (NOT an in-place delta): the CapabilityInvoker
|
||||
/// and its resolved options are bound to the child at spawn, so the only way a changed config takes
|
||||
/// effect is to rebuild the child (the respawn's Create call invalidates the stale cached pipelines).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ResilienceConfig_change_triggers_stop_plus_respawn()
|
||||
{
|
||||
var current = new Dictionary<string, DriverChildSnapshot>
|
||||
{
|
||||
// Same DriverType + same DriverConfig; only ResilienceConfig differs.
|
||||
["a"] = new("Modbus", "{\"host\":\"127.0.0.1\"}", ResilienceConfig: "{\"bulkheadMaxConcurrent\":8}"),
|
||||
};
|
||||
var target = new[] { Spec("a", resilience: "{\"bulkheadMaxConcurrent\":32}") };
|
||||
|
||||
var plan = DriverSpawnPlanner.Compute(current, target);
|
||||
|
||||
plan.ToStop.Single().ShouldBe("a");
|
||||
plan.ToSpawn.Single().DriverInstanceId.ShouldBe("a");
|
||||
plan.ToApplyDelta.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adding a ResilienceConfig where there was none (null → JSON) is also a respawn — the invoker
|
||||
/// was built with tier defaults and must be rebuilt to pick up the overrides.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Adding_ResilienceConfig_from_null_triggers_respawn()
|
||||
{
|
||||
var current = new Dictionary<string, DriverChildSnapshot>
|
||||
{
|
||||
["a"] = new("Modbus", "{\"host\":\"127.0.0.1\"}", ResilienceConfig: null),
|
||||
};
|
||||
var target = new[] { Spec("a", resilience: "{\"bulkheadMaxConcurrent\":16}") };
|
||||
|
||||
var plan = DriverSpawnPlanner.Compute(current, target);
|
||||
|
||||
plan.ToStop.Single().ShouldBe("a");
|
||||
plan.ToSpawn.Single().DriverInstanceId.ShouldBe("a");
|
||||
plan.ToApplyDelta.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A pure DriverConfig change with an UNCHANGED ResilienceConfig stays an in-place delta (no
|
||||
/// reconnect) — the resilience pipeline is untouched, so there's no reason to respawn.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void DriverConfig_change_with_unchanged_ResilienceConfig_stays_ApplyDelta()
|
||||
{
|
||||
var current = new Dictionary<string, DriverChildSnapshot>
|
||||
{
|
||||
["a"] = new("Modbus", "{\"host\":\"old\"}", ResilienceConfig: "{\"bulkheadMaxConcurrent\":16}"),
|
||||
};
|
||||
var target = new[] { Spec("a", config: "{\"host\":\"new\"}", resilience: "{\"bulkheadMaxConcurrent\":16}") };
|
||||
|
||||
var plan = DriverSpawnPlanner.Compute(current, target);
|
||||
|
||||
plan.ToApplyDelta.Single().DriverInstanceId.ShouldBe("a");
|
||||
plan.ToSpawn.ShouldBeEmpty();
|
||||
plan.ToStop.ShouldBeEmpty();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user