feat(drivers): periodic reconcile of desired vs actual subscriptions (#488)

Option A from the issue — the cheap consistency check, no interface change.

The desired ref set was re-applied on a Connected TRANSITION (`ResubscribeDesired`)
and on a deploy's `SetDesiredSubscriptions`, and nowhere else. A subscription lost
while the driver STAYED Connected — a subscribe that failed and was never retried,
a handle dropped down an error path — left the actor subscribed to nothing while
still reporting Healthy, indefinitely.

`ReconcileSubscription` now rides the existing 30 s `health-poll` tick: while
Connected, a driver holding a desired set but no live handle is re-subscribed.

Gated on `ISubscribable`. A driver that cannot subscribe at all can still be handed
a desired set, and without the guard it would be told `Subscribe` every tick and
fail every tick, forever.

This is a state-machine consistency check, NOT a probe — it can only see that this
actor holds no handle, never that a handle is present but dead server-side, because
`ISubscriptionHandle` is opaque. That is option B, and it stays deferred until a
real occurrence shows the handle outliving the subscription; it would need a
liveness member implemented across all eight drivers with different subscription
mechanics.

A redundant `Subscribe` is possible (a tick queued ahead of an already-queued one
observes the same no-handle state) and is harmless: `HandleSubscribeAsync` is a
`ReceiveAsync`, so no tick is processed mid-subscribe, and it drops any prior
subscription before establishing the new one. Suppressing it would need a
pending-subscribe flag threaded through every self-tell site — more state than the
race is worth.

`healthPollInterval` becomes an optional Props parameter, mirroring the existing
`rediscoverInterval` seam, so a test can drive the reconcile without waiting 30 s.

Three tests. The reconcile itself is proven by a stub whose FIRST subscribe throws:
nothing else would ever retry it, so a second attempt can only come from the
reconcile — disabling `ReconcileSubscription` turns it red. The two guard tests are
absence assertions and are bounded by a positive control rather than a sleep: a
counting health publisher makes ticks PROCESSED observable, so "still one subscribe"
means the reconcile ran and declined, not that the timer had not fired.

Explicitly not deploy-time drift (#486) — re-asserting an already-correct desired
set would not have helped there.

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs

@@ -170,6 +170,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
     /// <summary>Interval between bounded post-connect re-discovery passes. Production default 2s; tests
     /// inject a tiny value so the loop runs without real-time waits.</summary>
     private readonly TimeSpan _rediscoverInterval;
+    private readonly TimeSpan _healthPollInterval;
 
     /// <summary>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.</summary>
@@ -239,6 +240,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
     /// <param name="rediscoverDiscoverTimeout">Optional per-pass timeout for <see cref="ITagDiscovery.DiscoverAsync"/>; defaults to 30 seconds.</param>
     /// <param name="invoker">Optional Phase 6.1 resilience invoker wrapping this driver's capability
     /// calls; defaults to <see cref="NullDriverCapabilityInvoker"/> (pass-through) when not supplied.</param>
+    /// <param name="healthPollInterval">Optional period of the health-poll heartbeat, which also drives the
+    /// subscription reconcile (<see cref="ReconcileSubscription"/>); defaults to
+    /// <see cref="HealthPollInterval"/>. Exists so a test can drive the reconcile without waiting 30 s.</param>
     /// <returns>A <see cref="Akka.Actor.Props"/> instance configured to create the wrapped <see cref="DriverInstanceActor"/>.</returns>
     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));
 
     /// <summary>
     /// 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
     /// <param name="rediscoverDiscoverTimeout">Per-pass timeout for <see cref="ITagDiscovery.DiscoverAsync"/>; defaults to 30 seconds.</param>
     /// <param name="invoker">Phase 6.1 resilience invoker wrapping this driver's capability calls;
     /// defaults to <see cref="NullDriverCapabilityInvoker"/> (pass-through) when null.</param>
+    /// <param name="healthPollInterval">Period of the health-poll heartbeat, which also drives the
+    /// subscription reconcile; defaults to <see cref="HealthPollInterval"/> when null.</param>
     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<string, object?>("event", startStubbed ? "spawn_stub" : "spawn"),
             new KeyValuePair<string, object?>("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<SubscriptionFailed>(msg =>
             _log.Debug("DriverInstance {Id}: resubscribe reported failure: {Reason}", _driverInstanceId, msg.Reason));
-        Receive<HealthPollTick>(_ => PublishHealthSnapshot());
+        Receive<HealthPollTick>(_ =>
+        {
+            PublishHealthSnapshot();
+            ReconcileSubscription();
+        });
+    }
+
+    /// <summary>
+    ///     Periodic desired-vs-actual subscription reconcile (#488), riding the existing <c>health-poll</c>
+    ///     tick. While Connected, a driver that has a desired subscription set but no live handle is
+    ///     re-subscribed.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         The desired set is otherwise only (re)applied on a <b>Connected transition</b>
+    ///         (<see cref="ResubscribeDesired"/>) or on a deploy's <see cref="SetDesiredSubscriptions"/>.
+    ///         Nothing re-asserted it for a subscription that was lost while the driver <i>stayed</i>
+    ///         Connected — a failed subscribe whose retry never came, or a handle dropped down an error
+    ///         path. That is the hole this closes.
+    ///     </para>
+    ///     <para>
+    ///         <b>This is a state-machine consistency check, not a probe.</b> It can only see that this
+    ///         actor holds no handle; it cannot detect a handle that is present but dead server-side,
+    ///         because <c>ISubscriptionHandle</c> is opaque (a <c>DiagnosticId</c> string and nothing
+    ///         else). Detecting that needs a liveness member on <c>ISubscribable</c> implemented across
+    ///         all eight drivers, each with different subscription mechanics — deliberately deferred
+    ///         until a real occurrence shows the handle outliving the subscription.
+    ///     </para>
+    ///     <para>
+    ///         Gated on <see cref="ISubscribable"/>: a driver that cannot subscribe at all may still have
+    ///         been handed a desired set, and re-telling <c>Subscribe</c> to it every tick would fail
+    ///         every tick, forever.
+    ///     </para>
+    ///     <para>
+    ///         A redundant <c>Subscribe</c> is possible but harmless: a tick sitting in the mailbox ahead
+    ///         of an already-queued <c>Subscribe</c> observes the same "no handle" state and tells a
+    ///         second one. <see cref="HandleSubscribeAsync"/> is a <c>ReceiveAsync</c> (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.
+    ///     </para>
+    /// </remarks>
+    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()

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;
+
+/// <summary>
+///     Gitea #488 — the periodic desired-vs-actual subscription reconcile that rides
+///     <see cref="DriverInstanceActor"/>'s existing <c>health-poll</c> tick.
+/// </summary>
+/// <remarks>
+///     <para>
+///         The hole it closes: the desired ref set is re-applied on a <b>Connected transition</b> and on a
+///         deploy's <c>SetDesiredSubscriptions</c>, and nowhere else. A subscription lost while the driver
+///         <i>stayed</i> Connected — the shape reproduced here, a subscribe that failed and was never
+///         retried — left the actor permanently subscribed to nothing while reporting Healthy.
+///     </para>
+///     <para>
+///         All three tests drive a short <c>healthPollInterval</c> 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.
+///     </para>
+/// </remarks>
+public sealed class DriverInstanceActorSubscriptionReconcileTests : RuntimeActorTestBase
+{
+    private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(100);
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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" });
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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");
+    }
+
+    /// <summary>
+    ///     A driver that is not <see cref="ISubscribable"/> can still be handed a desired set. Reconciling
+    ///     it would tell <c>Subscribe</c> every tick and fail every tick, forever — so the reconcile is
+    ///     gated on the capability.
+    /// </summary>
+    [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);
+        });
+    }
+
+    /// <summary>Counts health-snapshot publishes, which is one per <c>health-poll</c> tick once the actor
+    /// is running — the observable that turns "time passed" into "ticks were processed".</summary>
+    private sealed class CountingHealthPublisher : IDriverHealthPublisher
+    {
+        private int _count;
+
+        /// <summary>Number of snapshots published so far.</summary>
+        public int Count => Volatile.Read(ref _count);
+
+        /// <inheritdoc />
+        public void Publish(string clusterId, string driverInstanceId, DriverHealth health, int errorCount5Min)
+            => Interlocked.Increment(ref _count);
+    }
+}

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).</summary>
     public bool UnsubscribeYields { get; set; }
 
+    /// <summary>
+    ///     How many of the first <see cref="SubscribeAsync"/> calls throw before one succeeds. Default 0
+    ///     (always succeed) leaves existing behaviour untouched.
+    /// </summary>
+    /// <remarks>
+    ///     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
+    ///     <c>_subscriptionHandle</c> is null and nothing else will ever re-assert it — there is no
+    ///     connect transition coming, and no deploy.
+    /// </remarks>
+    public int SubscribeFailuresBeforeSuccess { get; set; }
+
     /// <summary>Subscribes to the specified full references.</summary>
     /// <param name="fullReferences">The full references to subscribe to.</param>
     /// <param name="publishingInterval">The publishing interval.</param>
@@ -121,8 +133,13 @@ internal sealed class SubscribableStubDriver : StubDriver, ISubscribable
     public Task<ISubscriptionHandle> SubscribeAsync(
         IReadOnlyList<string> 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<ISubscriptionHandle>(_handle);
     }