using System.Collections;
using System.Collections.Concurrent;
using System.Reflection;
using NSubstitute;
using Opc.Ua;
using Opc.Ua.Client;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
///
/// Arch-review 2026-08-19 P2-4 (rollback half) — a batch subscribe whose
/// ApplyChanges throws must leave NO residue behind.
///
///
/// The gap this pins. CreateSubscriptionsCoreAsync registers each monitored item as
/// it places it (_callbacks, _itemShard, the shard's item count and the SDK
/// subscription's own item list) and only issues ApplyChanges once the whole chunk is
/// placed. A throw there — a dead session being the common cause — propagated with every one of
/// those registrations still in place and nothing left holding a handle to release them. The actor
/// sees the throw as a whole-chunk fault and re-issues the SAME batch on its backoff timer, so
/// each probe grew the bookkeeping again: unbounded shard growth and phantom shard occupancy.
///
///
///
/// The seam is reached with a data shard whose SDK was never created,
/// which is precisely how the SDK reports "you have no usable subscription": placement and
/// AddItem succeed, ApplyChangesAsync throws. Private state is read by reflection
/// rather than widened for the test.
///
///
[Trait("Category", "Unit")]
public class RealOpcUaClientBatchRollbackTests
{
private static object Field(object target, string name) =>
target.GetType().GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(target)!;
///
/// Gives the client a mocked session plus ONE pre-placed (never created) data shard, so a
/// batch subscribe places its items and then throws out of ApplyChanges.
///
private static Subscription ArrangeUncreatedShard(RealOpcUaClient client)
{
var session = Substitute.For();
session.NamespaceUris.Returns(new NamespaceTable());
typeof(RealOpcUaClient)
.GetField("_session", BindingFlags.Instance | BindingFlags.NonPublic)!
.SetValue(client, session);
// The parameterless ctor is obsolete in the current SDK in favour of a telemetry-context
// overload; nothing here publishes telemetry, and the alternative would be to hand-build a
// context purely to satisfy the signature.
#pragma warning disable CS0618
var subscription = new Subscription { DisplayName = "uncreated-shard" };
#pragma warning restore CS0618
var shardType = typeof(RealOpcUaClient).GetNestedType("DataShard", BindingFlags.NonPublic)!;
var shard = Activator.CreateInstance(
shardType,
BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic,
null, [subscription], null)!;
((IList)Field(client, "_dataShards")).Add(shard);
return subscription;
}
private static int ShardItemCount(RealOpcUaClient client)
{
var shard = ((IList)Field(client, "_dataShards"))[0]!;
return (int)shard.GetType().GetProperty("ItemCount")!.GetValue(shard)!;
}
[Fact]
public async Task ApplyChangesThrow_RollsBackEveryPartiallyCreatedItem()
{
var client = new RealOpcUaClient();
var subscription = ArrangeUncreatedShard(client);
var fault = await Record.ExceptionAsync(() => client.CreateSubscriptionsAsync(
["i=2258", "i=2259", "i=2260"],
(_, _, _, _) => { }));
// Positive control: the batch really did fail at ApplyChanges, not before placement.
Assert.NotNull(fault);
// Nothing registered survives the throw — pre-call state, exactly.
Assert.Empty((ConcurrentDictionary>)
Field(client, "_callbacks"));
Assert.Empty((ConcurrentDictionary)Field(client, "_itemShard"));
Assert.Empty((ConcurrentDictionary)Field(client, "_monitoredItems"));
Assert.Equal(0, ShardItemCount(client));
Assert.Empty(subscription.MonitoredItems);
}
[Fact]
public async Task RepeatedFailedBatches_DoNotGrowTheAdapterBookkeeping()
{
// The actual operational damage: the actor re-issues the same chunk on its backoff timer,
// so an un-rolled-back batch compounded on every probe.
var client = new RealOpcUaClient();
ArrangeUncreatedShard(client);
for (var attempt = 0; attempt < 3; attempt++)
{
var fault = await Record.ExceptionAsync(() => client.CreateSubscriptionsAsync(
["i=2258", "i=2259"],
(_, _, _, _) => { }));
Assert.NotNull(fault);
}
Assert.Empty((ConcurrentDictionary>)
Field(client, "_callbacks"));
Assert.Empty((ConcurrentDictionary)Field(client, "_itemShard"));
Assert.Equal(0, ShardItemCount(client));
}
}