diff --git a/docs/plans/2026-11-11-warm-redundancy-717-design.md b/docs/plans/2026-11-11-warm-redundancy-717-design.md index 25f3baf71..3c4db1847 100644 --- a/docs/plans/2026-11-11-warm-redundancy-717-design.md +++ b/docs/plans/2026-11-11-warm-redundancy-717-design.md @@ -323,3 +323,9 @@ Existence is not behaviour. Plan T0 re-proves, on 2.0.1 over the wire: - (d) the Publish/Republish choice with the new signatures; - (e) suppression and promotion refresh through the product's #854/#855 per-item paths, which did not exist when the 1.5 event spike ran. + +**Result (2026-11-13):** (a) and (b) are re-proven on 2.0.1. The metadata graph uses `IPredefinedNodeSubtypeReplacer` +in `OnNodeManagerStartedAsync`, which is simpler than the factory route. (c) and (d) **fail as written in §4.6**: +2.0.1 clones the cached message on every delivery and pools notification payloads, so neither object-keyed +provenance nor an object-keyed choice can work. An alternative mechanism passed the spike and awaits the owner's +decision. See `docs/plans/evidence/2026-11-13-717-sdk2/RESULTS.md`. diff --git a/docs/plans/2026-11-11-warm-redundancy-717-plan.md b/docs/plans/2026-11-11-warm-redundancy-717-plan.md index 9fc3f6c03..e9f39118d 100644 --- a/docs/plans/2026-11-11-warm-redundancy-717-plan.md +++ b/docs/plans/2026-11-11-warm-redundancy-717-plan.md @@ -42,6 +42,12 @@ The 1.5.378.106 spikes do not carry over (spec §10). Write throwaway loopback f **Exit:** a short results note per fixture, and an owner go/no-go. A "no" on (c) or (d) re-opens D4's mechanism. D4 itself stays an owner decision. +**T0 run 2026-11-13: (a) and (b) pass; (c) and (d) are no-go as written.** SDK 2.0.1 hands every Publish and Republish +a fresh clone of the cached message and pools the notification payloads, so provenance and the first-delivery choice +cannot be keyed by object. An alternative, keyed by `(SubscriptionId, SequenceNumber)` with a clone-surviving +provenance token, passed the same spike. It awaits the owner's go/no-go, and T1 onward are stopped until then. +Evidence: `docs/plans/evidence/2026-11-13-717-sdk2/RESULTS.md`. + ## T1: Configuration, validation, effective self URI **Files:** diff --git a/docs/plans/evidence/2026-11-13-717-sdk2/RESULTS.md b/docs/plans/evidence/2026-11-13-717-sdk2/RESULTS.md new file mode 100644 index 000000000..6c0ff0048 --- /dev/null +++ b/docs/plans/evidence/2026-11-13-717-sdk2/RESULTS.md @@ -0,0 +1,136 @@ +# #717 T0: SDK 2.0.1 hook re-proof results (2026-11-13) + +Plan task T0 of `docs/plans/2026-11-11-warm-redundancy-717-plan.md`. A throwaway loopback server +(`Warm717Sdk2Spike.cs` in this folder: a `StandardServer` with a `CustomNodeManager2`, as the product uses) was +driven over a real OPC UA session on the pinned `OPCFoundation.NetStandard.Opc.Ua.*` **2.0.1**. The client used raw +`CreateSubscription` / `Publish` / `Republish` / `SetPublishingMode` service calls, so each step is deterministic. +SDK source references are to the `2.0.1` tag of `OPCFoundation/UA-.NETStandard`. + +Run: 11 of 11 passed (16 s). The (c)/(d) tests run twice: `V2=false` pins the failure of design §4.6 as written, +and `V2=true` pins the alternative described below. + +## Verdict + +| Fixture | Result | Gate | +|---|---|---| +| a. Metadata graph | **Pass**, through a different hook than planned | go | +| b. Read projection coverage | **Pass** | go | +| c. Provenance | **Fail as designed.** The custom item installs, but object-keyed provenance never reaches a delivered notification | **no-go on §4.6 as written** | +| d. Publish/Republish choice | **Fail as designed.** Every delivery is a fresh clone, so a choice keyed by object never matches | **no-go on §4.6 as written** | +| e. Events | Not exercised. The stamp/deliver split is product code, not an SDK hook | n/a | + +The plan's T0 exit rule applies: "A 'no' on (c) or (d) re-opens D4's mechanism." Implementation (T1 onward) is +stopped pending the owner's go/no-go on the alternative mechanism below. D4 itself (strict first delivery) is unchanged. + +## a. Metadata graph: pass + +- `Server.ServerRedundancy` is promoted to `NonTransparentRedundancyType` (2039) by + `IPredefinedNodeSubtypeReplacer.ReplacePredefinedInstanceSubtypeAsync`. The interface is new in 2.0.1, and the + `DiagnosticsNodeManager` (an `AsyncCustomNodeManager`) implements it. The call goes in `OnNodeManagerStartedAsync`, + which the product already overrides, so no `CreateMainNodeManagerFactory` override is needed. The well-known child + NodeId `i=11314` is passed in, so the passive-predefined-node problem from the 1.5 spike (c7791) does not arise. + This is the same route the SDK's own `Opc.Ua.Redundancy.Server` package uses (`ServerRedundancyController.ApplyAsync`). + That package is not referenced, and it publishes peers only, whereas decision #155 wants self first. +- Over the wire, in Warm: `i=3709` = 2 (Warm); `HasTypeDefinition` of `i=2296` = `i=2039`; `ServerRedundancy` + `HasProperty` → `i=11314`; `TranslateBrowsePaths(Server/ServerRedundancy/ServerUriArray)` = `i=11314`; + `i=11314` = `[self, peer]`; a Write to it is refused; `Server.ServiceLevel` reads 1 at boot. +- In None: `i=3709` = 0, and the type definition stays `ServerRedundancyType` (2034). +- **Finding:** in None (and so also today's Unset), the predefined `i=11314` survives as an orphan. It is readable by + NodeId (Good, empty `String[]`), but it is not referenced from `ServerRedundancy` (browse lists only + `RedundancySupport` and `RedundantServerArray`), and TranslateBrowsePaths returns `BadNoMatch`. `docs/Redundancy.md` + says `i=11314` is "absent". That is true for browse and false for a direct Read. Spec §3 wants it absent for None. + Deleting the orphan in None is a small product step. + +## b. Read projection: pass + +An override of `BaseVariableState.ReadValueAttribute` (the product's `EquipmentVariableState` already has one) that +runs the base read first and then returns `BadOutOfService` with a null value covers every path: + +- Read, including an IndexRange read of an array tag. The `ServerTimestamp` is set; non-Value attributes stay Good. +- The initial sample on create, with queue size 1 and with queue size 10. +- The change fan-out while projected: a new stored value produces no Good notification. +- Sampling → Reporting. +- `Server.ResendData` (`i=12873`): the resent value is projected. +- AggregateFilter (Average, 200 ms): no Good aggregate over projected input. + +A non-ordinary variable in the same manager and `i=2267` stay Good. Turning the projection off and touching the node +delivers the latest stored value Good, with no new driver value. In 2.0.1, `ReadAttributeAsync` and the +`MonitoredNode2` sampling paths fall back to the synchronous `ReadAttribute` → `ReadValueAttribute` when no +`OnReadValueAsync`/`OnSimpleReadValueAsync` is set (`NodeState.cs:4556`, `BaseVariableState.cs:2063`), and the product +sets neither. A queued value with `ignoreFilters: true` reaches an item with an absolute-deadband filter in both +directions, which is the T8 demotion/promotion queue. + +## c./d. Provenance and first-delivery choice: fail as designed + +**The custom sampled item can be installed.** The planned 2.0.1 creation hook (`MonitoredItemFactory` / +`MonitoredItemCreateDecision`) exists only on `AsyncCustomNodeManager` (`AsyncCustomNodeManager.cs:8550`). Its +`CustomNodeManager2` counterpart (`ICustomMonitoredItemManager`) is `internal`. But `CustomNodeManager2.m_monitoredItemManager` +is protected and assignable, and `MonitoredNodeMonitoredItemManager` is public and not sealed. A subclass that +re-implements `IMonitoredItemManager.CreateMonitoredItem` creates the item, and keeps the SDK's +`is MonitoredNodeMonitoredItemManager` checks and its internal lifecycle interface intact. That is the 1.5 +"manager-subclass route". The spike confirms the custom item is the one in use. + +**Object identity does not survive delivery.** In 2.0.1, `SentMessageQueue` hands every delivery a deep clone of the +cached message: a new message's first send (`SentMessageQueue.cs:310`), a prebuilt queued message (`:204`) and +Republish (`:357`) all go through `CoreUtils.Clone`. Notification payloads are pooled: `ReuseNotificationPayloads` +(`:529`) returns them to `MonitoredItemNotificationActivator` on acknowledge, eviction and overflow, and +`ResetForReuse` → `Initialize()` resets only `ClientHandle` and `Value`. So: + +- A provenance table keyed by the `MonitoredItemNotification` the item created **never** matches a delivered + notification. The spike recorded 0 object hits. +- A first-delivery ledger keyed by the message object never matches either, because each Publish and Republish of the + same sequence number is a different object. +- Worse, object-keyed provenance would go **stale**: a pooled object stamped for an ordinary tag can be handed back + out to any other item's notification. + +With `V2=false`, the spike shows Good ordinary-tag values delivered after demotion, both for samples queued before +demotion and for messages prebuilt before it. + +Two further 2.0.1 facts change the test matrix: + +- A Republish of a prebuilt but not-yet-sent sequence number is refused with `BadMessageNotAvailable`. "First + delivery through Republish" (spec §4.6 item 2) cannot occur for queued messages. +- An anonymous session cannot `TransferSubscriptions` another anonymous session's subscription + (`BadUserAccessDenied`). The transfer leg was not exercised. + +## Alternative mechanism proven in the spike (`V2=true`), for the owner's go/no-go + +This keeps the D4 semantics and spec §4.6's structure: a custom item stamps at build time, and an override of +`PublishAsync`/`RepublishAsync` decides at first delivery. It changes the two keys. + +1. **Ledger keyed by `(SubscriptionId, SequenceNumber)`, not by object.** The first delivery of a sequence number + records Active or Standby. Every later delivery of that number, including Republish, applies the recorded + decision to the fresh clone the SDK returns. A Standby decision projects the clone, which is our own object, so no + cached SDK object is touched. Entries are pruned against the `AvailableSequenceNumbers` that Publish returns. The + ledger must be bounded per subscription and for dead subscriptions. +2. **Provenance token carried on `MonitoredItemNotification.Message`.** This is a public, non-encoded helper + reference that the SDK's `MemberwiseClone` copies into every clone. The custom item sets it, after the base + per-value publish has applied the timestamp filter, to a token holding the notification's `ClientHandle` and + final `DataValue`. At delivery a notification counts as ordinary-tag only if its token matches its own + `ClientHandle` and `DataValue`. A stale token left on a pooled object (`Initialize()` does not clear the helper) + therefore fails the match. + +Spike evidence with `V2=true`: + +- Good samples queued while Active, a ClientHandle change, then demotion: the first Publish carries + `BadOutOfService`/null, a non-ordinary item in the same message stays Good, and Republish returns the projected payload. +- Messages prebuilt while Active (`maxNotificationsPerPublish=1`), then demotion: the next Publish is projected, and + Republish returns the recorded projected choice. +- Good delivered while Active, then demotion: Republish returns the original Good. After an acknowledge, Republish + returns `BadMessageNotAvailable`. + +Limits the owner should weigh: + +- The token re-purposes an SDK helper property. The SDK server never reads it (no server use in the 2.0.1 source), + and the SDK version is pinned. A contract test would pin the clone-copies-helper behaviour so an SDK bump fails loudly. +- Stale-token rejection is by fingerprint. A false match needs the same `ClientHandle` *and* an identical `DataValue` + (value, status and both timestamps) on a pooled object. That is improbable but not impossible by construction. +- A durable-subscription restore (out of scope, spec §8) and a TransferSubscriptions under one identity are still + unexercised. + +Other ways to close the gap, not spiked: + +- (i) Migrate `OtOpcUaNodeManager` to `AsyncCustomNodeManager` to get the supported creation hook. This is a very + large change, and it still does not fix the clone/pool problem. +- (ii) Drop D4's queued/prebuilt guarantee and rely on the item-level projection alone. That is an owner decision on + D4 itself. diff --git a/docs/plans/evidence/2026-11-13-717-sdk2/Warm717Sdk2Spike.cs b/docs/plans/evidence/2026-11-13-717-sdk2/Warm717Sdk2Spike.cs new file mode 100644 index 000000000..541a2dc85 --- /dev/null +++ b/docs/plans/evidence/2026-11-13-717-sdk2/Warm717Sdk2Spike.cs @@ -0,0 +1,835 @@ +// Gitea #717 T0 — THROWAWAY SDK 2.0.1 re-proof spike (2026-11-13). Not product code; not in any build. +// To rerun: copy into tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ and run +// dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests --filter "FullyQualifiedName~Warm717" +// A minimal StandardServer + CustomNodeManager2 that exercises, over a real loopback session, the hook shapes the +// Warm design depends on: (a) the redundancy metadata graph, (b) per-tag read projection on every value path, +// (c) per-notification provenance through a custom sampled item, (d) the Publish/Republish first-delivery choice. +// SpikeState.V2 = false is design §4.6 as written (object-keyed); V2 = true is the alternative in RESULTS.md. +using System.Runtime.CompilerServices; +using Microsoft.Extensions.Logging; +using Opc.Ua; +using Opc.Ua.Client; +using Opc.Ua.Server; +using Opc.Ua.Configuration; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +public sealed class Warm717Sdk2Spike +{ + private const string Ns = "urn:spike717"; + + internal sealed class SpikeState + { + public volatile bool Projected; + public volatile bool StandbyDelivery; + public readonly object ChoiceLock = new(); + public readonly ConditionalWeakTable Choices = new(); + public int Rewritten; + public RedundancySupport Mode = RedundancySupport.None; + public bool V2; + public readonly Dictionary<(uint Sub, uint Seq), bool> Ledger = new(); + public int ObjectHits; + public string SelfUri = ""; + public string[] Peers = []; + } + + internal sealed record Stamp(NodeId NodeId, uint AttributeId); + + // Rides MonitoredItemNotification.Message, a non-encoded helper reference that the SDK's MemberwiseClone copies. + internal sealed class ProvenanceToken(uint clientHandle, DataValue value, Stamp stamp) : NotificationMessage + { + public uint ClientHandle { get; } = clientHandle; + public DataValue Value { get; } = value; + public Stamp Stamp { get; } = stamp; + public bool Matches(MonitoredItemNotification n) => n.ClientHandle == ClientHandle && n.Value.Equals(Value); + } + + internal static readonly ConditionalWeakTable Provenance = new(); + internal static int StampCount, ChooseCalls, DcnSeen, Unstamped; + + private sealed class SpikeVariable(NodeState? parent, SpikeState state) : BaseDataVariableState(parent) + { + protected override ServiceResult ReadValueAttribute(ISystemContext context, NumericRange indexRange, + QualifiedName dataEncoding, ref Variant value, ref DateTimeUtc sourceTimestamp) + { + var r = base.ReadValueAttribute(context, indexRange, dataEncoding, ref value, ref sourceTimestamp); + if (ServiceResult.IsBad(r) || !state.Projected) return r; + value = Variant.Null; + sourceTimestamp = DateTimeUtc.MinValue; + return StatusCodes.BadOutOfService; + } + } + + private sealed class StampingItem(IServerInternal server, IAsyncNodeManager nm, NodeHandle handle, uint subId, uint id, + MonitoredItemCreateRequest req, DiagnosticsMasks dm, TimestampsToReturn ttr, MonitoringFilter? filter, Opc.Ua.Range? eu, + double sampling, uint queue, bool durable) + : Opc.Ua.Server.MonitoredItem(server, nm, handle, subId, id, req.ItemToMonitor, dm, ttr, req.MonitoringMode, + req.RequestedParameters.ClientHandle, filter, filter, eu, sampling, queue, req.RequestedParameters.DiscardOldest, 0, durable) + { + public override bool Publish(OperationContext context, Queue notifications, + Queue diagnostics, uint maxNotificationsPerPublish, ILogger logger) + { + var local = new Queue(); + var localDiag = new Queue(); + var more = base.Publish(context, local, localDiag, maxNotificationsPerPublish, logger); + var stamp = new Stamp(NodeId, AttributeId); + while (local.Count > 0) + { + var n = local.Dequeue(); + Provenance.AddOrUpdate(n, stamp); + n.Message = new ProvenanceToken(n.ClientHandle, n.Value, stamp); + Interlocked.Increment(ref StampCount); + notifications.Enqueue(n); + diagnostics.Enqueue(localDiag.Dequeue()); + } + return more; + } + } + + // Re-implements IMonitoredItemManager.CreateMonitoredItem (interface re-implementation), so CustomNodeManager2's + // `m_monitoredItemManager is MonitoredNodeMonitoredItemManager` checks and the internal lifecycle interface still hold. + private sealed class ProvenanceItemManager(IAsyncNodeManager nm, IServerInternal server) + : MonitoredNodeMonitoredItemManager(nm, server), IMonitoredItemManager + { + ISampledDataChangeMonitoredItem IMonitoredItemManager.CreateMonitoredItem(IServerInternal server, IAsyncNodeManager nodeManager, + ServerSystemContext context, NodeHandle handle, uint subscriptionId, double publishingInterval, + DiagnosticsMasks diagnosticsMasks, TimestampsToReturn timestampsToReturn, MonitoredItemCreateRequest itemToCreate, + Opc.Ua.Range euRange, MonitoringFilter filterToUse, double samplingInterval, uint revisedQueueSize, bool createDurable, + MonitoredItemIdFactory monitoredItemIdFactory, Func addNodeToComponentCache) + { + if (handle.Node is not SpikeVariable || itemToCreate.ItemToMonitor.AttributeId != Attributes.Value) + { + return CreateMonitoredItem(server, nodeManager, context, handle, subscriptionId, publishingInterval, diagnosticsMasks, + timestampsToReturn, itemToCreate, euRange, filterToUse, samplingInterval, revisedQueueSize, createDurable, + monitoredItemIdFactory, addNodeToComponentCache); + } + + var cachedNode = addNodeToComponentCache(context, handle, handle.Node); + if (!MonitoredNodes.TryGetValue(handle.Node.NodeId, out var monitoredNode)) + { + MonitoredNodes[handle.Node.NodeId] = monitoredNode = new MonitoredNode2(nodeManager, server, cachedNode, + nodeManager.IsMultipleEventConsumerNode(cachedNode.NodeId)); + } + handle.Node = monitoredNode.Node; + handle.MonitoredNode = monitoredNode; + uint id; + do { id = monitoredItemIdFactory.GetNextId(); } while (!MonitoredItems.TryAdd(id, null!)); + var item = new StampingItem(server, nodeManager, handle, subscriptionId, id, itemToCreate, diagnosticsMasks, + timestampsToReturn, filterToUse, euRange, samplingInterval, revisedQueueSize, createDurable); + monitoredNode.Add((IDataChangeMonitoredItem2)item); + MonitoredItems[id] = item; + return item; + } + } + + private sealed class SpikeNodeManager : CustomNodeManager2 + { + private readonly SpikeState _state; + public SpikeVariable? Tag; + public BaseDataVariableState? Meta; + public SpikeVariable? Arr; + + public SpikeNodeManager(IServerInternal server, ApplicationConfiguration config, SpikeState state) + : base(server, config, Ns) + { + _state = state; + m_monitoredItemManager.Dispose(); + m_monitoredItemManager = new ProvenanceItemManager(this.ToAsyncNodeManager(), server); + } + + public override void CreateAddressSpace(IDictionary> externalReferences) + { + lock (Lock) + { + base.CreateAddressSpace(externalReferences); + var folder = new FolderState(null) + { + NodeId = new NodeId("root", NamespaceIndex), BrowseName = new QualifiedName("root", NamespaceIndex), + DisplayName = new LocalizedText("root"), TypeDefinitionId = ObjectTypeIds.FolderType, + }; + folder.AddReference(ReferenceTypeIds.Organizes, true, ObjectIds.ObjectsFolder); + if (!externalReferences.TryGetValue(ObjectIds.ObjectsFolder, out var refs)) + externalReferences[ObjectIds.ObjectsFolder] = refs = new List(); + refs.Add(new NodeStateReference(ReferenceTypeIds.Organizes, false, folder.NodeId)); + Tag = Var(folder, new SpikeVariable(folder, _state), "Tag", DataTypeIds.Double, ValueRanks.Scalar, 5.0); + Arr = Var(folder, new SpikeVariable(folder, _state), "Arr", DataTypeIds.Int32, ValueRanks.OneDimension, new Variant(new[] { 1, 2, 3 })); + Meta = Var(folder, new BaseDataVariableState(folder), "Meta", DataTypeIds.Double, ValueRanks.Scalar, 7.0); + AddPredefinedNode(SystemContext, folder); + } + } + + private T Var(FolderState folder, T v, string name, NodeId dataType, int rank, Variant value) where T : BaseDataVariableState + { + v.NodeId = new NodeId(name, NamespaceIndex); + v.BrowseName = new QualifiedName(name, NamespaceIndex); + v.DisplayName = new LocalizedText(name); + v.TypeDefinitionId = VariableTypeIds.BaseDataVariableType; + v.ReferenceTypeId = ReferenceTypeIds.Organizes; + v.DataType = dataType; + v.ValueRank = rank; + v.AccessLevel = v.UserAccessLevel = AccessLevels.CurrentReadOrWrite; + v.Value = value; + v.StatusCode = StatusCodes.Good; + v.Timestamp = DateTimeUtc.Now; + folder.AddChild(v); + return v; + } + + public void Set(BaseDataVariableState v, Variant value) + { + lock (Lock) + { + v.Value = value; + v.StatusCode = StatusCodes.Good; + v.Timestamp = DateTimeUtc.Now; + v.ClearChangeMasks(SystemContext, false); + } + } + + public void Touch(BaseDataVariableState v) + { + lock (Lock) + { + v.Timestamp = DateTimeUtc.Now; + v.ClearChangeMasks(SystemContext, false); + } + } + + public IEnumerable Items() => MonitoredItems.Values; + + public int QueueToAll(Func make) + { + var n = 0; + foreach (var item in MonitoredItems.Values) + { + if (item is StampingItem s) { s.QueueValue(make(), null, true); n++; } + } + return n; + } + } + + private sealed class SpikeServer(SpikeState state) : StandardServer(DefaultTelemetry.Create(_ => { })) + { + public SpikeNodeManager? NodeManager; + + protected override MasterNodeManager CreateMasterNodeManager(IServerInternal server, ApplicationConfiguration configuration) + { + NodeManager = new SpikeNodeManager(server, configuration, state); + return new MasterNodeManager(server, configuration, null, NodeManager); + } + + protected override async ValueTask OnNodeManagerStartedAsync(IServerInternal server, CancellationToken cancellationToken = default) + { + await base.OnNodeManagerStartedAsync(server, cancellationToken); + var so = server.ServerObject; + var ctx = server.DefaultSystemContext; + so.ServerRedundancy!.RedundancySupport!.Value = state.Mode; + if (state.Mode == RedundancySupport.Warm) + { + var replacer = (IPredefinedNodeSubtypeReplacer)server.DiagnosticsNodeManager; + var subtype = ctx.CreateInstanceOfNonTransparentRedundancyType(); + var replaced = (NonTransparentRedundancyState)await replacer.ReplacePredefinedInstanceSubtypeAsync(ctx, + so.ServerRedundancy, subtype, + new Dictionary { [new QualifiedName(BrowseNames.ServerUriArray, 0)] = VariableIds.Server_ServerRedundancy_ServerUriArray }, + n => so.ServerRedundancy = (ServerRedundancyState)n, cancellationToken); + replaced.ServerUriArray!.Value = (ArrayOf)new[] { state.SelfUri }.Concat(state.Peers).ToArray(); + replaced.ServerUriArray.AccessLevel = replaced.ServerUriArray.UserAccessLevel = AccessLevels.CurrentRead; + replaced.ClearChangeMasks(ctx, true); + so.ServiceLevel!.Value = (byte)1; + so.ServiceLevel.ClearChangeMasks(ctx, false); + } + } + + public override async ValueTask PublishAsync(SecureChannelContext secureChannelContext, RequestHeader? requestHeader, + ArrayOf subscriptionAcknowledgements, RequestLifetime requestLifetime) + { + var response = await base.PublishAsync(secureChannelContext, requestHeader, subscriptionAcknowledgements, requestLifetime); + response.NotificationMessage = state.V2 ? Choose2(response.SubscriptionId, response.NotificationMessage) : Choose(response.NotificationMessage); + return response; + } + + public override async ValueTask RepublishAsync(SecureChannelContext secureChannelContext, RequestHeader? requestHeader, + uint subscriptionId, uint retransmitSequenceNumber, RequestLifetime requestLifetime) + { + var response = await base.RepublishAsync(secureChannelContext, requestHeader, subscriptionId, retransmitSequenceNumber, requestLifetime); + response.NotificationMessage = state.V2 ? Choose2(subscriptionId, response.NotificationMessage) : Choose(response.NotificationMessage); + return response; + } + + private NotificationMessage Choose(NotificationMessage message) + { + Interlocked.Increment(ref ChooseCalls); + if (message is null || message.NotificationData.Count == 0) return message!; + lock (state.ChoiceLock) + { + if (state.Choices.TryGetValue(message, out var chosen)) return chosen; + chosen = state.StandbyDelivery ? Project(message) : message; + state.Choices.AddOrUpdate(message, chosen); + return chosen; + } + } + + // V2: one recorded choice per (subscription, sequence number); the SDK hands every delivery a fresh clone. + private NotificationMessage Choose2(uint sub, NotificationMessage message) + { + if (message is null || message.NotificationData.Count == 0) return message!; + bool standby; + lock (state.ChoiceLock) + { + if (!state.Ledger.TryGetValue((sub, message.SequenceNumber), out standby)) + state.Ledger[(sub, message.SequenceNumber)] = standby = state.StandbyDelivery; + } + return standby ? Project(message) : message; + } + + private NotificationMessage Project(NotificationMessage message) + { + var data = new List(); + var changed = false; + for (var i = 0; i < message.NotificationData.Count; i++) + { + var eo = message.NotificationData[i]; + if (eo.TryGetValue(out var dcn)) + { + Interlocked.Increment(ref DcnSeen); + for (var j = 0; j < dcn.MonitoredItems.Count; j++) if (!Provenance.TryGetValue(dcn.MonitoredItems[j], out _)) Interlocked.Increment(ref Unstamped); + var items = new List(); + var any = false; + for (var j = 0; j < dcn.MonitoredItems.Count; j++) + { + var n = dcn.MonitoredItems[j]; + if (Provenance.TryGetValue(n, out _)) Interlocked.Increment(ref state.ObjectHits); + var ordinary = state.V2 ? n.Message is ProvenanceToken t && t.Matches(n) : Provenance.TryGetValue(n, out _); + if (ordinary && StatusCode.IsNotBad(n.Value.StatusCode)) + { + items.Add(new MonitoredItemNotification + { + ClientHandle = n.ClientHandle, + Value = new DataValue(Variant.Null, StatusCodes.BadOutOfService, DateTimeUtc.MinValue, n.Value.ServerTimestamp), + }); + any = true; + Interlocked.Increment(ref state.Rewritten); + } + else items.Add(n); + } + if (any) + { + changed = true; + data.Add(new ExtensionObject(new DataChangeNotification { MonitoredItems = items, DiagnosticInfos = dcn.DiagnosticInfos })); + continue; + } + } + data.Add(eo); + } + if (!changed) return message; + return new NotificationMessage { SequenceNumber = message.SequenceNumber, PublishTime = message.PublishTime, NotificationData = data }; + } + } + + // ---------------------------------------------------------------- harness + private sealed class Rig : IAsyncDisposable + { + public required SpikeServer Server; + public required ApplicationInstance App; + public required SpikeState State; + public required Opc.Ua.Client.ISession Session; + public required string Pki; + public required string ClientPki; + public required ApplicationConfiguration ClientConfig; + public ushort Ns; + public NodeId Tag => new("Tag", Ns); + public NodeId Arr => new("Arr", Ns); + public NodeId Meta => new("Meta", Ns); + public SpikeNodeManager Nm => Server.NodeManager!; + public SessionClient Raw => (SessionClient)Session; + + public async ValueTask DisposeAsync() + { + try { await Session.CloseAsync(2000, true, CancellationToken.None); } catch { } + Session.Dispose(); + try { await Server.StopAsync(CancellationToken.None); } catch { } + Server.Dispose(); + foreach (var d in new[] { Pki, ClientPki }) try { Directory.Delete(d, true); } catch { } + } + } + + private static async Task StartAsync(SpikeState state, CancellationToken ct) + { + var pki = Path.Combine(Path.GetTempPath(), $"spike717-{Guid.NewGuid():N}"); + var clientPki = TestClientSecurity.AllocatePkiRoot(); + for (var attempt = 0; ; attempt++) + { + var port = OpcUaHostStart.AllocateFreePort(); + state.SelfUri = $"urn:spike717:self:{port}"; + var app = new ApplicationInstance(DefaultTelemetry.Create(_ => { })) { ApplicationName = "Spike717", ApplicationType = ApplicationType.Server }; + var config = new ApplicationConfiguration(DefaultTelemetry.Create(_ => { })) + { + ApplicationName = "Spike717", ApplicationUri = state.SelfUri, ApplicationType = ApplicationType.Server, + ServerConfiguration = new ServerConfiguration + { + BaseAddresses = [$"opc.tcp://localhost:{port}/Spike"], + SecurityPolicies = [new ServerSecurityPolicy { SecurityMode = MessageSecurityMode.None, SecurityPolicyUri = SecurityPolicies.None }], + UserTokenPolicies = [new UserTokenPolicy(UserTokenType.Anonymous)], + MaxFailedAuthenticationAttempts = 0, + }, + SecurityConfiguration = new SecurityConfiguration + { + ApplicationCertificate = new CertificateIdentifier { StoreType = "Directory", StorePath = Path.Combine(pki, "own"), SubjectName = "CN=Spike717" }, + TrustedIssuerCertificates = new CertificateTrustList { StoreType = "Directory", StorePath = Path.Combine(pki, "issuer") }, + TrustedPeerCertificates = new CertificateTrustList { StoreType = "Directory", StorePath = Path.Combine(pki, "trusted") }, + RejectedCertificateStore = new CertificateTrustList { StoreType = "Directory", StorePath = Path.Combine(pki, "rejected") }, + AutoAcceptUntrustedCertificates = true, + }, + TransportQuotas = new TransportQuotas { OperationTimeout = 10_000 }, + ClientConfiguration = new ClientConfiguration(), + }; + app.ApplicationConfiguration = config; + await config.ValidateAsync(ApplicationType.Server, ct); + await app.CheckApplicationInstanceCertificatesAsync(false, null, ct); + var server = new SpikeServer(state); + try + { + await app.StartAsync(server, ct); + } + catch (Exception ex) when (attempt < 4 && OpcUaHostStart.IsBindFailure(ex)) + { + server.Dispose(); + continue; + } + + var cc = new ApplicationConfiguration(DefaultTelemetry.Create(_ => { })) + { + ApplicationName = "Spike717Client", ApplicationUri = $"urn:spike717client:{Guid.NewGuid():N}", + ApplicationType = ApplicationType.Client, SecurityConfiguration = TestClientSecurity.Build(clientPki), + ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 30_000 }, + TransportQuotas = new TransportQuotas { OperationTimeout = 10_000 }, + }; + await cc.ValidateAsync(ApplicationType.Client, ct); + var ep = await CoreClientUtils.SelectEndpointAsync(cc, $"opc.tcp://localhost:{port}/Spike", false, DefaultTelemetry.Create(_ => { }), ct); + var session = await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(cc, + new ConfiguredEndpoint(null, ep!, EndpointConfiguration.Create(cc)), false, false, "spike", 30_000, + new UserIdentity(new AnonymousIdentityToken()), default, ct); + var rig = new Rig { Server = server, App = app, State = state, Session = session, Pki = pki, ClientPki = clientPki, ClientConfig = cc }; + rig.Ns = (ushort)session.NamespaceUris.GetIndex(Ns); + return rig; + } + } + + private static RequestHeader H() => new() { Timestamp = DateTimeUtc.Now, TimeoutHint = 10_000 }; + + private static async Task CreateSubAsync(Rig rig, uint maxPerPublish, CancellationToken ct, bool enabled = true) + { + var r = await rig.Raw.CreateSubscriptionAsync(H(), 50, 300, 10, maxPerPublish, enabled, 0, ct); + return r.SubscriptionId; + } + + private static async Task AddItemAsync(Rig rig, uint sub, NodeId node, uint handle, uint queue, CancellationToken ct, + MonitoringFilter? filter = null, MonitoringMode mode = MonitoringMode.Reporting, string? indexRange = null) + { + var req = new MonitoredItemCreateRequest + { + ItemToMonitor = new ReadValueId { NodeId = node, AttributeId = Attributes.Value, IndexRange = indexRange }, + MonitoringMode = mode, + RequestedParameters = new MonitoringParameters + { + ClientHandle = handle, SamplingInterval = 0, QueueSize = queue, DiscardOldest = true, + Filter = filter is null ? ExtensionObject.Null : new ExtensionObject(filter), + }, + }; + var r = await rig.Raw.CreateMonitoredItemsAsync(H(), sub, TimestampsToReturn.Both, new[] { req }, ct); + StatusCode.IsGood(r.Results[0].StatusCode).ShouldBeTrue($"create {node}: {r.Results[0].StatusCode}"); + return r.Results[0].MonitoredItemId; + } + + private static List<(uint Handle, DataValue Value)> DataChanges(NotificationMessage m) + { + var list = new List<(uint, DataValue)>(); + for (var i = 0; i < m.NotificationData.Count; i++) + { + if (m.NotificationData[i].TryGetValue(out var d)) + for (var j = 0; j < d.MonitoredItems.Count; j++) list.Add((d.MonitoredItems[j].ClientHandle, d.MonitoredItems[j].Value)); + } + return list; + } + + private static async Task PublishAsync(Rig rig, CancellationToken ct, params SubscriptionAcknowledgement[] acks) + => await rig.Raw.PublishAsync(H(), acks, ct); + + // Publishes until a message with notifications arrives (skipping keep-alives). + private static async Task PublishDataAsync(Rig rig, CancellationToken ct) + { + for (var i = 0; i < 40; i++) + { + var r = await PublishAsync(rig, ct); + if (r.NotificationMessage.NotificationData.Count > 0) return r; + } + throw new InvalidOperationException("no data message"); + } + + private static CancellationToken Deadline(int seconds = 60) + { + var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + cts.CancelAfter(TimeSpan.FromSeconds(seconds)); + return cts.Token; + } + + // ---------------------------------------------------------------- (a) metadata + [Fact] + public async Task A_warm_metadata_graph_over_the_wire() + { + var ct = Deadline(); + var state = new SpikeState { Mode = RedundancySupport.Warm, Peers = ["urn:spike717:peer"] }; + await using var rig = await StartAsync(state, ct); + + var support = await rig.Session.ReadValueAsync(VariableIds.Server_ServerRedundancy_RedundancySupport, ct); + support.StatusCode.ShouldBe(StatusCodes.Good); + Convert.ToInt32(support.ClrValue()).ShouldBe((int)RedundancySupport.Warm); + + var uris = await rig.Session.ReadValueAsync(VariableIds.Server_ServerRedundancy_ServerUriArray, ct); + uris.StatusCode.ShouldBe(StatusCodes.Good); + ((string[])uris.ClrValue()!).ShouldBe(new[] { state.SelfUri, "urn:spike717:peer" }); + + var browse = await rig.Raw.BrowseAsync(H(), null, 0, new[] + { + new BrowseDescription { NodeId = ObjectIds.Server_ServerRedundancy, BrowseDirection = BrowseDirection.Forward, + ReferenceTypeId = ReferenceTypeIds.HasTypeDefinition, IncludeSubtypes = true, ResultMask = (uint)BrowseResultMask.All }, + new BrowseDescription { NodeId = ObjectIds.Server_ServerRedundancy, BrowseDirection = BrowseDirection.Forward, + ReferenceTypeId = ReferenceTypeIds.HasProperty, IncludeSubtypes = true, ResultMask = (uint)BrowseResultMask.All }, + }, ct); + browse.Results[0].References.Count.ShouldBe(1); + browse.Results[0].References[0].NodeId.ShouldBe(new ExpandedNodeId(ObjectTypeIds.NonTransparentRedundancyType)); + var props = Enumerable.Range(0, browse.Results[1].References.Count).Select(i => browse.Results[1].References[i]).ToList(); + props.ShouldContain(r => r.NodeId == new ExpandedNodeId(VariableIds.Server_ServerRedundancy_ServerUriArray)); + + var tr = await rig.Raw.TranslateBrowsePathsToNodeIdsAsync(H(), new[] + { + new BrowsePath + { + StartingNode = ObjectIds.Server, + RelativePath = new RelativePath { Elements = new[] + { + new RelativePathElement { ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, TargetName = new QualifiedName(BrowseNames.ServerRedundancy) }, + new RelativePathElement { ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, TargetName = new QualifiedName(BrowseNames.ServerUriArray) }, + } }, + }, + }, ct); + tr.Results[0].StatusCode.ShouldBe(StatusCodes.Good); + tr.Results[0].Targets[0].TargetId.ShouldBe(new ExpandedNodeId(VariableIds.Server_ServerRedundancy_ServerUriArray)); + + var w = await rig.Raw.WriteAsync(H(), new[] + { + new WriteValue { NodeId = VariableIds.Server_ServerRedundancy_ServerUriArray, AttributeId = Attributes.Value, + Value = new DataValue(new Variant(new[] { "x" })) }, + }, ct); + StatusCode.IsBad(w.Results[0]).ShouldBeTrue($"ServerUriArray write must be refused, got {w.Results[0]}"); + + var sl = await rig.Session.ReadValueAsync(VariableIds.Server_ServiceLevel, ct); + Convert.ToByte(sl.ClrValue()).ShouldBe((byte)1); + } + + [Fact] + public async Task A_none_metadata_graph_over_the_wire() + { + var ct = Deadline(); + await using var rig = await StartAsync(new SpikeState { Mode = RedundancySupport.None }, ct); + var support = await rig.Session.ReadValueAsync(VariableIds.Server_ServerRedundancy_RedundancySupport, ct); + Convert.ToInt32(support.ClrValue()).ShouldBe((int)RedundancySupport.None); + var uris = await rig.Raw.ReadAsync(H(), 0, TimestampsToReturn.Both, new[] + { + new ReadValueId { NodeId = VariableIds.Server_ServerRedundancy_ServerUriArray, AttributeId = Attributes.Value }, + }, ct); + var noneBrowse = await rig.Raw.BrowseAsync(H(), null, 0, new[] + { + new BrowseDescription { NodeId = ObjectIds.Server_ServerRedundancy, BrowseDirection = BrowseDirection.Forward, + ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, ResultMask = (uint)BrowseResultMask.All }, + }, ct); + var children = Enumerable.Range(0, noneBrowse.Results[0].References.Count).Select(i => noneBrowse.Results[0].References[i].BrowseName.Name).ToList(); + var tr = await rig.Raw.TranslateBrowsePathsToNodeIdsAsync(H(), new[] + { + new BrowsePath { StartingNode = ObjectIds.Server_ServerRedundancy, RelativePath = new RelativePath { Elements = new[] + { new RelativePathElement { ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, TargetName = new QualifiedName(BrowseNames.ServerUriArray) } } } }, + }, ct); + // FINDING (2.0.1): the predefined i=11314 survives as an orphan — readable by NodeId (Good, empty String[]) but + // not referenced from Server.ServerRedundancy (browse lists only RedundancySupport + RedundantServerArray). + uris.Results[0].StatusCode.ShouldBe(StatusCodes.Good); + children.ShouldBe(new[] { BrowseNames.RedundancySupport, BrowseNames.RedundantServerArray }, ignoreOrder: true); + tr.Results[0].StatusCode.ShouldBe(StatusCodes.BadNoMatch); + var browse = await rig.Raw.BrowseAsync(H(), null, 0, new[] + { + new BrowseDescription { NodeId = ObjectIds.Server_ServerRedundancy, BrowseDirection = BrowseDirection.Forward, + ReferenceTypeId = ReferenceTypeIds.HasTypeDefinition, IncludeSubtypes = true, ResultMask = (uint)BrowseResultMask.All }, + }, ct); + browse.Results[0].References[0].NodeId.ShouldBe(new ExpandedNodeId(ObjectTypeIds.ServerRedundancyType)); + } + + // ---------------------------------------------------------------- (b) read projection coverage + [Fact] + public async Task B_every_value_path_reaches_the_projection() + { + var ct = Deadline(); + var state = new SpikeState(); + await using var rig = await StartAsync(state, ct); + + // Read: Good, then projected, then Good again (acquisition kept the stored value). + (await rig.Session.ReadValueAsync(rig.Tag, ct)).StatusCode.ShouldBe(StatusCodes.Good); + state.Projected = true; + var projected = (await rig.Raw.ReadAsync(H(), 0, TimestampsToReturn.Both, new[] { new ReadValueId { NodeId = rig.Tag, AttributeId = Attributes.Value } }, ct)).Results[0]; + projected.StatusCode.ShouldBe(StatusCodes.BadOutOfService); + projected.WrappedValue.IsNull.ShouldBeTrue(); + (projected.ServerTimestamp != DateTimeUtc.MinValue).ShouldBeTrue(); + // IndexRange read + Meta (not ordinary) stays Good + ServiceLevel stays Good. + var ranged = await rig.Raw.ReadAsync(H(), 0, TimestampsToReturn.Both, new[] + { + new ReadValueId { NodeId = rig.Arr, AttributeId = Attributes.Value, IndexRange = "1" }, + new ReadValueId { NodeId = rig.Meta, AttributeId = Attributes.Value }, + new ReadValueId { NodeId = VariableIds.Server_ServiceLevel, AttributeId = Attributes.Value }, + new ReadValueId { NodeId = rig.Tag, AttributeId = Attributes.DisplayName }, + }, ct); + ranged.Results[0].StatusCode.ShouldBe(StatusCodes.BadOutOfService); + ranged.Results[1].StatusCode.ShouldBe(StatusCodes.Good); + ranged.Results[2].StatusCode.ShouldBe(StatusCodes.Good); + ranged.Results[3].StatusCode.ShouldBe(StatusCodes.Good); + + // Initial sample on create (queue 1 and queue 10), plus an AggregateFilter item, plus a Sampling-mode item. + var sub = await CreateSubAsync(rig, 0, ct); + await AddItemAsync(rig, sub, rig.Tag, 1, 1, ct); + await AddItemAsync(rig, sub, rig.Tag, 2, 10, ct); + await AddItemAsync(rig, sub, rig.Meta, 3, 1, ct); + var aggId = await AddItemAsync(rig, sub, rig.Tag, 4, 10, ct, new AggregateFilter + { + AggregateType = ObjectIds.AggregateFunction_Average, ProcessingInterval = 200, StartTime = DateTimeUtc.Now, + AggregateConfiguration = new AggregateConfiguration { UseServerCapabilitiesDefaults = true }, + }); + var samplingId = await AddItemAsync(rig, sub, rig.Tag, 5, 1, ct, mode: MonitoringMode.Sampling); + var first = DataChanges((await PublishDataAsync(rig, ct)).NotificationMessage); + first.Where(x => x.Handle is 1 or 2).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService); + first.Where(x => x.Handle is 1 or 2).Count().ShouldBe(2); + first.ShouldContain(x => x.Handle == 3 && x.Value.StatusCode == StatusCodes.Good); + + // Change fan-out while projected: a new stored value produces no Good notification. + rig.Nm.Set(rig.Nm.Tag!, 6.0); + // Sampling -> Reporting while projected: projected. + await rig.Raw.SetMonitoringModeAsync(H(), sub, MonitoringMode.Reporting, new[] { samplingId }, ct); + var collected = new List<(uint Handle, DataValue Value)>(); + var deadline = DateTime.UtcNow.AddSeconds(3); + while (DateTime.UtcNow < deadline) + { + var r = await PublishAsync(rig, ct); + collected.AddRange(DataChanges(r.NotificationMessage)); + } + collected.Where(x => x.Handle is 1 or 2 or 5).ShouldAllBe(x => x.Value.StatusCode != StatusCodes.Good, + "no Good ordinary-tag notification while projected"); + collected.Where(x => x.Handle == 4).ShouldAllBe(x => StatusCode.IsBad(x.Value.StatusCode) || x.Value.StatusCode == StatusCodes.GoodNoData, + "aggregate over projected input is not Good data"); + + // ResendData (Server.ResendData method, i=12873) while projected: the resent value is projected. + var call = await rig.Raw.CallAsync(H(), new[] + { + new CallMethodRequest { ObjectId = ObjectIds.Server, MethodId = MethodIds.Server_ResendData, InputArguments = new[] { new Variant(sub) } }, + }, ct); + call.Results[0].StatusCode.ShouldBe(StatusCodes.Good); + var resent = DataChanges((await PublishDataAsync(rig, ct)).NotificationMessage); + resent.Where(x => x.Handle is 1 or 2).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService); + resent.ShouldContain(x => x.Handle == 1); + + // Un-project: a touch delivers the latest stored value Good without a new driver value. + state.Projected = false; + rig.Nm.Touch(rig.Nm.Tag!); + var back = DataChanges((await PublishDataAsync(rig, ct)).NotificationMessage); + back.ShouldContain(x => x.Handle == 1 && x.Value.StatusCode == StatusCodes.Good && Equals(x.Value.WrappedValue.AsBoxedObject(), 6.0)); + _ = aggId; + } + + // ---------------------------------------------------------------- (c)+(d) provenance and first-delivery choice + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CD_queued_good_samples_then_demotion_first_delivery_is_projected(bool v2) + { + var ct = Deadline(); + var state = new SpikeState { V2 = v2 }; + await using var rig = await StartAsync(state, ct); + var sub = await CreateSubAsync(rig, 0, ct); + await AddItemAsync(rig, sub, rig.Tag, 1, 10, ct); + await AddItemAsync(rig, sub, rig.Meta, 2, 10, ct); + await AddItemAsync(rig, sub, VariableIds.Server_ServiceLevel, 3, 1, ct); + DataChanges((await PublishDataAsync(rig, ct)).NotificationMessage).Count.ShouldBeGreaterThan(0); + + // Queue Good samples with publishing disabled, then demote (D4 boundary), then publish. + await rig.Raw.SetPublishingModeAsync(H(), false, new[] { sub }, ct); + rig.Nm.Set(rig.Nm.Tag!, 10.0); + rig.Nm.Set(rig.Nm.Tag!, 11.0); + rig.Nm.Set(rig.Nm.Meta!, 12.0); + // Modify the client handle between queue and publish: provenance is by object, not handle. + var modify = await rig.Raw.ModifyMonitoredItemsAsync(H(), sub, TimestampsToReturn.Both, new[] + { + new MonitoredItemModifyRequest { MonitoredItemId = 1, RequestedParameters = new MonitoringParameters { ClientHandle = 99, SamplingInterval = 0, QueueSize = 10, DiscardOldest = true } }, + }, ct); + _ = modify; + state.StandbyDelivery = true; + state.Projected = true; + await rig.Raw.SetPublishingModeAsync(H(), true, new[] { sub }, ct); + var msg = await PublishDataAsync(rig, ct); + var changes = DataChanges(msg.NotificationMessage); + changes.Where(x => x.Handle is 1 or 99).ShouldNotBeEmpty(); + var diag = "items: " + string.Join(",", rig.Nm.Items().Select(i => i.GetType().Name)) + " rewritten=" + state.Rewritten + " stamps=" + StampCount + " choose=" + ChooseCalls + " dcn=" + DcnSeen + " unstamped=" + Unstamped + " standby=" + state.StandbyDelivery + " objectHits=" + state.ObjectHits; + rig.Nm.Items().OfType().Count().ShouldBe(1, "the custom sampled item IS installed (manager re-implementation)"); + if (!v2) + { + // FINDING (2.0.1): design §4.6 as written — provenance and the choice keyed by object — never matches a + // delivered object (SentMessageQueue hands every delivery a CoreUtils.Clone), so Good leaks after demotion. + state.ObjectHits.ShouldBe(0, diag); + changes.Where(x => x.Handle is 1 or 99).ShouldContain(x => x.Value.StatusCode == StatusCodes.Good, diag); + return; + } + changes.Where(x => x.Handle is 1 or 99).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService && x.Value.WrappedValue.IsNull, + "Good ordinary-tag samples queued before demotion are first-delivered projected " + diag); + changes.ShouldContain(x => x.Handle == 2 && x.Value.StatusCode == StatusCodes.Good, "non-ordinary item untouched"); + + // Republish of the projected message returns the same projected payload. + var again = await rig.Raw.RepublishAsync(H(), sub, msg.NotificationMessage.SequenceNumber, ct); + DataChanges(again.NotificationMessage).Where(x => x.Handle is 1 or 99).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService); + state.Rewritten.ShouldBeGreaterThan(0); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CD_good_delivered_while_active_republishes_good_after_demotion_and_ack_drops_it(bool v2) + { + var ct = Deadline(); + var state = new SpikeState { V2 = v2 }; + await using var rig = await StartAsync(state, ct); + var sub = await CreateSubAsync(rig, 0, ct); + await AddItemAsync(rig, sub, rig.Tag, 1, 10, ct); + var first = await PublishDataAsync(rig, ct); + DataChanges(first.NotificationMessage).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.Good); + + state.StandbyDelivery = true; + state.Projected = true; + var again = await rig.Raw.RepublishAsync(H(), sub, first.NotificationMessage.SequenceNumber, ct); + DataChanges(again.NotificationMessage).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.Good, + "Republish of a message first delivered while Active returns the original (D4 boundary)"); + + await PublishAsync(rig, ct, new SubscriptionAcknowledgement { SubscriptionId = sub, SequenceNumber = first.NotificationMessage.SequenceNumber }); + var ex = await Should.ThrowAsync(async () => + await rig.Raw.RepublishAsync(H(), sub, first.NotificationMessage.SequenceNumber, ct)); + ex.StatusCode.ShouldBe(StatusCodes.BadMessageNotAvailable); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CD_prebuilt_messages_then_demotion_first_delivery_projected_and_through_republish(bool v2) + { + var ct = Deadline(); + var state = new SpikeState { V2 = v2 }; + await using var rig = await StartAsync(state, ct); + // maxNotificationsPerPublish = 1 so one Publish builds several messages and queues the rest (prebuilt). + var sub = await CreateSubAsync(rig, 1, ct); + await AddItemAsync(rig, sub, rig.Tag, 1, 10, ct); + DataChanges((await PublishDataAsync(rig, ct)).NotificationMessage).Count.ShouldBe(1); + + await rig.Raw.SetPublishingModeAsync(H(), false, new[] { sub }, ct); + rig.Nm.Set(rig.Nm.Tag!, 20.0); + rig.Nm.Set(rig.Nm.Tag!, 21.0); + rig.Nm.Set(rig.Nm.Tag!, 22.0); + await rig.Raw.SetPublishingModeAsync(H(), true, new[] { sub }, ct); + var a = await PublishDataAsync(rig, ct); // builds 3 messages while Active; first delivered Good + DataChanges(a.NotificationMessage).Single().Value.StatusCode.ShouldBe(StatusCodes.Good); + a.AvailableSequenceNumbers.Count.ShouldBeGreaterThanOrEqualTo(1); + + state.StandbyDelivery = true; // demotion after the messages were prebuilt + var next = a.NotificationMessage.SequenceNumber + 1; + // First delivery THROUGH Republish of a prebuilt-but-unsent message (if the SDK allows it), else via Publish. + NotificationMessage? viaRepublish = null; + try { viaRepublish = (await rig.Raw.RepublishAsync(H(), sub, next, ct)).NotificationMessage; } + catch (ServiceResultException e) when (e.StatusCode == StatusCodes.BadMessageNotAvailable) { } + var b = await PublishDataAsync(rig, ct); + TestContext.Current.TestOutputHelper?.WriteLine($"v2={v2} first-delivery-through-Republish of a prebuilt message: {(viaRepublish is null ? "refused BadMessageNotAvailable" : "served")}"); + if (!v2) + { + DataChanges(b.NotificationMessage).ShouldContain(x => x.Value.StatusCode == StatusCodes.Good, "FINDING: object-keyed choice leaks Good"); + return; + } + DataChanges(b.NotificationMessage).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService, + "a message built while Active but first delivered after demotion is projected"); + if (viaRepublish is not null) + { + DataChanges(viaRepublish).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService); + b.NotificationMessage.SequenceNumber.ShouldBe(next); + ReferenceEquals(viaRepublish, b.NotificationMessage).ShouldBeFalse(); // decoded separately, same payload + } + var bRe = await rig.Raw.RepublishAsync(H(), sub, b.NotificationMessage.SequenceNumber, ct); + DataChanges(bRe.NotificationMessage).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService, + "Republish returns the recorded projected choice"); + + } + + [Fact] + public async Task C_provenance_survives_transfer_and_the_cached_message_is_not_mutated() + { + var ct = Deadline(); + var state = new SpikeState(); + await using var rig = await StartAsync(state, ct); + var sub = await CreateSubAsync(rig, 0, ct); + await AddItemAsync(rig, sub, rig.Tag, 1, 10, ct); + var first = await PublishDataAsync(rig, ct); + DataChanges(first.NotificationMessage).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.Good); + + // A second session takes the subscription over; the items (and their provenance) move with it. + var cc = rig.ClientConfig; + var second = await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(cc, + rig.Session.ConfiguredEndpoint, false, false, "spike2", 30_000, new UserIdentity(new AnonymousIdentityToken()), default, ct); + try + { + var raw2 = (SessionClient)second; + var t = await raw2.TransferSubscriptionsAsync(H(), new[] { sub }, false, ct); + // FINDING (2.0.1): an anonymous session may not take over another anonymous session's subscription, so the + // transfer leg cannot run on this anonymous-only spike server. Recorded, not exercised. + t.Results[0].StatusCode.ShouldBe(StatusCodes.BadUserAccessDenied); + if (t.Results[0].StatusCode == StatusCodes.BadUserAccessDenied) return; + rig.Nm.Set(rig.Nm.Tag!, 30.0); + state.StandbyDelivery = true; + PublishResponse? got = null; + for (var i = 0; i < 40 && got is null; i++) + { + var r = await raw2.PublishAsync(H(), default, ct); + if (DataChanges(r.NotificationMessage).Count > 0) got = r; + } + got.ShouldNotBeNull(); + DataChanges(got!.NotificationMessage).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService); + // the first message (delivered Active) is still Good on republish — the cached object was never mutated. + var re = await raw2.RepublishAsync(H(), sub, first.NotificationMessage.SequenceNumber, ct); + DataChanges(re.NotificationMessage).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.Good); + } + finally + { + try { await second.CloseAsync(2000, true, ct); } catch { } + second.Dispose(); + } + } + + [Fact] + public async Task C_demotion_queue_with_ignore_filters_reaches_deadband_items() + { + var ct = Deadline(); + var state = new SpikeState(); + await using var rig = await StartAsync(state, ct); + var sub = await CreateSubAsync(rig, 0, ct); + await AddItemAsync(rig, sub, rig.Tag, 1, 1, ct, new DataChangeFilter { Trigger = DataChangeTrigger.StatusValue, DeadbandType = (uint)DeadbandType.Absolute, DeadbandValue = 1000 }); + await PublishDataAsync(rig, ct); + state.Projected = true; + rig.Nm.QueueToAll(() => new DataValue(Variant.Null, StatusCodes.BadOutOfService, DateTimeUtc.MinValue, DateTimeUtc.Now)).ShouldBe(1); + DataChanges((await PublishDataAsync(rig, ct)).NotificationMessage).Single().Value.StatusCode.ShouldBe(StatusCodes.BadOutOfService); + state.Projected = false; + var n = 0; + foreach (var item in new[] { rig.Nm.Tag! }) { n++; } + rig.Nm.QueueToAll(() => { var dv = new DataValue(); return new DataValue(new Variant(5.0), StatusCodes.Good, DateTimeUtc.Now, DateTimeUtc.Now); }); + var promo = DataChanges((await PublishDataAsync(rig, ct)).NotificationMessage).Single(); + promo.Value.StatusCode.ShouldBe(StatusCodes.Good, "promotion value passes a deadband filter when queued ignoring filters"); + _ = n; + } +}