diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 3373c182..c0ca7f97 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -170,6 +170,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Interval between bounded post-connect re-discovery passes. Production default 2s; tests /// inject a tiny value so the loop runs without real-time waits. private readonly TimeSpan _rediscoverInterval; + private readonly TimeSpan _healthPollInterval; /// Cap on the number of post-connect re-discovery passes — a backstop so a never-stabilising /// (or perpetually-empty) discovered set cannot spin the loop forever. Production default 15. @@ -239,6 +240,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Optional per-pass timeout for ; defaults to 30 seconds. /// Optional Phase 6.1 resilience invoker wrapping this driver's capability /// calls; defaults to (pass-through) when not supplied. + /// Optional period of the health-poll heartbeat, which also drives the + /// subscription reconcile (); defaults to + /// . Exists so a test can drive the reconcile without waiting 30 s. /// A instance configured to create the wrapped . public static Props Props( IDriver driver, @@ -249,7 +253,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers TimeSpan? rediscoverInterval = null, int rediscoverMaxAttempts = DefaultRediscoverMaxAttempts, TimeSpan? rediscoverDiscoverTimeout = null, - IDriverCapabilityInvoker? invoker = null) => + IDriverCapabilityInvoker? invoker = null, + TimeSpan? healthPollInterval = null) => Akka.Actor.Props.Create(() => new DriverInstanceActor( driver, reconnectInterval ?? DefaultReconnectInterval, @@ -259,7 +264,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers rediscoverInterval, rediscoverMaxAttempts, rediscoverDiscoverTimeout, - invoker)); + invoker, + healthPollInterval)); /// /// Returns true when the driver should boot in DEV-STUB mode based on host platform and @@ -293,6 +299,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Per-pass timeout for ; defaults to 30 seconds. /// Phase 6.1 resilience invoker wrapping this driver's capability calls; /// defaults to (pass-through) when null. + /// Period of the health-poll heartbeat, which also drives the + /// subscription reconcile; defaults to when null. public DriverInstanceActor( IDriver driver, TimeSpan reconnectInterval, @@ -302,7 +310,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers TimeSpan? rediscoverInterval = null, int rediscoverMaxAttempts = DefaultRediscoverMaxAttempts, TimeSpan? rediscoverDiscoverTimeout = null, - IDriverCapabilityInvoker? invoker = null) + IDriverCapabilityInvoker? invoker = null, + TimeSpan? healthPollInterval = null) { _driver = driver; _invoker = invoker ?? NullDriverCapabilityInvoker.Instance; @@ -313,6 +322,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers _rediscoverInterval = rediscoverInterval ?? DefaultRediscoverInterval; _rediscoverMaxAttempts = rediscoverMaxAttempts; _rediscoverDiscoverTimeout = rediscoverDiscoverTimeout ?? DefaultRediscoverDiscoverTimeout; + _healthPollInterval = healthPollInterval ?? HealthPollInterval; OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, new KeyValuePair("event", startStubbed ? "spawn_stub" : "spawn"), new KeyValuePair("driver_type", driver.DriverType)); @@ -335,7 +345,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // actor starts, before any state transition fires. Also start the periodic heartbeat so // long-lived Healthy drivers keep their snapshot fresh for newly-joined SignalR clients. PublishHealthSnapshot(); - Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, HealthPollInterval); + Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, _healthPollInterval); } private void Stubbed() @@ -489,7 +499,58 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // to Self (HandleSubscribeAsync already logged the underlying cause). Swallow it so it doesn't dead-letter. Receive(msg => _log.Debug("DriverInstance {Id}: resubscribe reported failure: {Reason}", _driverInstanceId, msg.Reason)); - Receive(_ => PublishHealthSnapshot()); + Receive(_ => + { + PublishHealthSnapshot(); + ReconcileSubscription(); + }); + } + + /// + /// Periodic desired-vs-actual subscription reconcile (#488), riding the existing health-poll + /// tick. While Connected, a driver that has a desired subscription set but no live handle is + /// re-subscribed. + /// + /// + /// + /// The desired set is otherwise only (re)applied on a Connected transition + /// () or on a deploy's . + /// Nothing re-asserted it for a subscription that was lost while the driver stayed + /// Connected — a failed subscribe whose retry never came, or a handle dropped down an error + /// path. That is the hole this closes. + /// + /// + /// This is a state-machine consistency check, not a probe. It can only see that this + /// actor holds no handle; it cannot detect a handle that is present but dead server-side, + /// because ISubscriptionHandle is opaque (a DiagnosticId string and nothing + /// else). Detecting that needs a liveness member on ISubscribable implemented across + /// all eight drivers, each with different subscription mechanics — deliberately deferred + /// until a real occurrence shows the handle outliving the subscription. + /// + /// + /// Gated on : a driver that cannot subscribe at all may still have + /// been handed a desired set, and re-telling Subscribe to it every tick would fail + /// every tick, forever. + /// + /// + /// A redundant Subscribe is possible but harmless: a tick sitting in the mailbox ahead + /// of an already-queued Subscribe observes the same "no handle" state and tells a + /// second one. is a ReceiveAsync (so no tick can be + /// processed mid-subscribe) and drops any prior subscription before establishing the new one, + /// so the duplicate costs one extra round-trip and converges. Suppressing it would need a + /// pending-subscribe flag threaded through every self-tell site — more state than the race + /// is worth. + /// + /// + private void ReconcileSubscription() + { + if (_driver is not ISubscribable) return; + if (_desiredRefs.Count == 0 || _subscriptionHandle is not null) return; + + _log.Info( + "DriverInstance {Id}: connected with {Count} desired subscription ref(s) but no live subscription handle — re-subscribing (periodic reconcile)", + _driverInstanceId, _desiredRefs.Count); + Self.Tell(new Subscribe(_desiredRefs, _desiredInterval)); } private void Reconnecting() diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs new file mode 100644 index 00000000..1a901650 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs @@ -0,0 +1,122 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Gitea #488 — the periodic desired-vs-actual subscription reconcile that rides +/// 's existing health-poll tick. +/// +/// +/// +/// The hole it closes: the desired ref set is re-applied on a Connected transition and on a +/// deploy's SetDesiredSubscriptions, and nowhere else. A subscription lost while the driver +/// stayed Connected — the shape reproduced here, a subscribe that failed and was never +/// retried — left the actor permanently subscribed to nothing while reporting Healthy. +/// +/// +/// All three tests drive a short healthPollInterval rather than the production 30 s. The +/// two absence assertions are bounded by a positive control (a counted number of ticks provably +/// processed), never by a bare sleep. +/// +/// +public sealed class DriverInstanceActorSubscriptionReconcileTests : RuntimeActorTestBase +{ + private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(100); + + /// + /// The reconcile itself: the first subscribe fails, leaving the actor Connected with a desired set + /// and no handle. Nothing else would ever retry it — no reconnect, no deploy — so a second + /// subscribe can only come from the periodic reconcile. + /// + [Fact] + public void A_lost_subscription_is_re_established_while_still_connected() + { + var driver = new SubscribableStubDriver { SubscribeFailuresBeforeSuccess = 1 }; + var actor = Sys.ActorOf(DriverInstanceActor.Props(driver, healthPollInterval: FastPoll)); + + actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( + new[] { "tag-a" }, TimeSpan.FromMilliseconds(100))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + // The connect-path subscribe runs and throws, so the handle stays null. + AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget); + + // Only the reconcile can produce this second attempt — and it must succeed this time. + AwaitCondition(() => driver.SubscribeCount >= 2, PresenceBudget); + driver.LastSubscribedRefs.ShouldBe(new[] { "tag-a" }); + } + + /// + /// A healthy subscription is left alone — the reconcile must not re-subscribe on every tick, which + /// would churn the backend every 30 s in production for no reason. + /// + [Fact] + public void A_healthy_subscription_is_not_re_subscribed_on_every_tick() + { + var publisher = new CountingHealthPublisher(); + var driver = new SubscribableStubDriver(); + var actor = Sys.ActorOf(DriverInstanceActor.Props( + driver, healthPublisher: publisher, healthPollInterval: FastPoll)); + + actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( + new[] { "tag-a" }, TimeSpan.FromMilliseconds(100))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget); + + // POSITIVE CONTROL for the absence below. Every health-poll tick publishes a snapshot, so the + // publisher count is a direct observation of ticks PROCESSED. Waiting for several of them is what + // makes "SubscribeCount is still 1" mean "the reconcile ran and declined" rather than "the timer + // had not fired yet" — the latter would pass instantly and prove nothing. + var before = publisher.Count; + AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget); + + driver.SubscribeCount.ShouldBe(1, + "the handle is live, so the reconcile must leave it alone"); + } + + /// + /// A driver that is not can still be handed a desired set. Reconciling + /// it would tell Subscribe every tick and fail every tick, forever — so the reconcile is + /// gated on the capability. + /// + [Fact] + public void A_non_subscribable_driver_is_never_reconciled() + { + var publisher = new CountingHealthPublisher(); + var driver = new StubDriver(); // IDriver only — no ISubscribable + var actor = Sys.ActorOf(DriverInstanceActor.Props( + driver, healthPublisher: publisher, healthPollInterval: FastPoll)); + + actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( + new[] { "tag-a" }, TimeSpan.FromMilliseconds(100))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + // Same positive control: assert the reconcile stays silent across several PROCESSED ticks. + // EventFilter bounds the absence to this block, and the reconcile announces itself at Info — + // the first test above is the proof that this message does appear when the reconcile fires. + EventFilter.Info(contains: "periodic reconcile").Expect(0, () => + { + var before = publisher.Count; + AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget); + }); + } + + /// Counts health-snapshot publishes, which is one per health-poll tick once the actor + /// is running — the observable that turns "time passed" into "ticks were processed". + private sealed class CountingHealthPublisher : IDriverHealthPublisher + { + private int _count; + + /// Number of snapshots published so far. + public int Count => Volatile.Read(ref _count); + + /// + public void Publish(string clusterId, string driverInstanceId, DriverHealth health, int errorCount5Min) + => Interlocked.Increment(ref _count); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs index 9e83fb46..2ab4213d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs @@ -114,6 +114,18 @@ internal sealed class SubscribableStubDriver : StubDriver, ISubscribable /// synchronously-completed stub task hides (the continuation otherwise runs inline). public bool UnsubscribeYields { get; set; } + /// + /// How many of the first calls throw before one succeeds. Default 0 + /// (always succeed) leaves existing behaviour untouched. + /// + /// + /// This is how a test reaches the state the #488 reconcile exists for: the actor is Connected and + /// holds a desired ref set, but the subscribe that should have established the handle failed, so + /// _subscriptionHandle is null and nothing else will ever re-assert it — there is no + /// connect transition coming, and no deploy. + /// + public int SubscribeFailuresBeforeSuccess { get; set; } + /// Subscribes to the specified full references. /// The full references to subscribe to. /// The publishing interval. @@ -121,8 +133,13 @@ internal sealed class SubscribableStubDriver : StubDriver, ISubscribable public Task SubscribeAsync( IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) { - Interlocked.Increment(ref SubscribeCount); + var attempt = Interlocked.Increment(ref SubscribeCount); LastSubscribedRefs = fullReferences; + if (attempt <= SubscribeFailuresBeforeSuccess) + { + throw new InvalidOperationException($"stub subscribe failure #{attempt}"); + } + return Task.FromResult(_handle); }