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);
}
}