From 4769044f88e63408704bdb682731784d7c78f0aa Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 12 Aug 2026 02:56:17 -0400 Subject: [PATCH] feat(galaxy): handle the replay-gap sentinel and sparse-array values MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the two follow-ups left open by the MxGateway 0.2.0 bump. Reading the gateway's own contract first changed what each of them is, and one of my earlier claims was simply wrong -- both corrections are recorded below. Replay gap ---------- A sentinel MxEvent carries family = UNSPECIFIED, so it fell through EventPump's family switch into the default drop: we silently ignored a documented MUST ("you missed events -- discard local state and re-snapshot"). It is now intercepted and answered by re-issuing SubscribeBulk for every tracked subscription, since MXAccess delivers an initial value per item on subscribe -- re-subscribing IS the re-snapshot. It does not restart the event pump (the stream carrying the sentinel is healthy, and the handler runs on that stream's own read loop, so disposing the pump from there would deadlock on itself), it does not block the read loop, and re-entry is gated so a burst collapses to one pass instead of a subscribe storm per sentinel. Handled on the READ loop, not in the dispatch loop, because everything past that point is subject to newest-dropped channel pressure -- and "you missed events" is the one signal whose loss is unrecoverable. Pinned by a test that fills the channel and still expects the sentinel. ⚠️ This cannot fire in the shipped configuration, and I did not pretend otherwise or paper over it by enabling resume. The gateway emits the sentinel only for a stream opened with after_worker_sequence > 0; GatewayGalaxySubscriber passes 0, which the gateway treats as "fresh stream, not a resume -- plain attach, no replay, no sentinel". Enabling resume is a larger and genuinely delicate change: the sequence is only meaningful within ONE gateway session, our reconnect recreates the session, and replaying a remembered sequence against a fresh session sets the server's live watermark ABOVE every event it is about to send -- silencing the stream instead of recovering it. That deserves its own change with a live gate, which I cannot run from here. Sparse arrays -- correcting the last commit message --------------------------------------------------- I wrote that the bump "makes handling it possible" on the decode path. That is wrong: MxSparseArray is documented write-only, the gateway expands it before forwarding, and "the worker never receives or produces it". So the decode path is unreachable by contract and the write path has no use for it either -- it is a wire compaction for a WHOLE-array write, not a partial update (unmentioned indices are RESET, deliberately, because the gateway cannot read-modify-write without racing the provider). It is therefore the wrong primitive for an OPC UA IndexRange write, which expects untouched elements preserved -- and WriteRequest carries no IndexRange in the first place. It is decoded anyway, as contract defence: the alternative if a future gateway ever emits one is a silent null published under Good quality. Fill semantics mirror the gateway's own SparseArrayExpander exactly (epoch for Time, empty string, Integer widening to long[] when any element is int64) so a value that round-trips through a write reads back as the same CLR array. Also: the decoder is now total by contract, since it runs inside the dispatch loop where an escaping exception kills dispatch for the life of the pump. A malformed payload degrades instead of throwing. A real defect found in the same code ------------------------------------ A stream the GATEWAY completes -- a closed or evicted session -- ends the RPC without an error, so nothing threw, the supervisor was never notified, and the pump exited silently. GetHealth() kept reporting Healthy, which also means the #527 comms-loss watchdog could never trip, because it requires a non-Healthy state. The driver went permanently deaf with no signal anywhere. This one is live, unlike the two above. Now reported as a fault; the falsification control is the existing CleanShutdown_DoesNotInvokeOnStreamFault, since our own disposal also ends the stream and must NOT be reported. Value kinds we cannot represent are now counted and warned once rather than published as an indistinguishable null. The status is deliberately NOT downgraded to Bad: nothing can reach that branch today, and changing what an OPC UA client sees for a live shape is not a change to make without a gate. The counter exists so the next person has evidence instead of a guess. Verified: 330 Galaxy tests green (12 new), plus Galaxy.Browser 21, Runtime 558, Core 351; full solution builds 0 errors with the warning census unchanged. Every new test was falsified first -- disabling the gap interception reds 3 of 4 (the throwing-callback test passes vacuously when the callback never runs, by construction), disabling sparse decode plus the representability signal reds all 7 decoder tests, and both were restored and re-verified green. Docs: the Galaxy.Performance invariant was stated as received = dispatched + dropped + in-flight, which this breaks -- a sentinel is received but never enters the channel -- so it now reads + replay_gaps. Galaxy.md gains the three operational notes; DriverLifecycle.md records that "reaches a non-Healthy state" is an obligation a driver can silently fail to meet, which is the general shape of the clean-end bug. --- docs/DriverLifecycle.md | 10 + docs/drivers/Galaxy.md | 4 + docs/v2/Galaxy.Performance.md | 25 +- .../GalaxyDriver.cs | 110 +++++++- .../Runtime/EventPump.cs | 122 +++++++- .../Runtime/MxValueDecoder.cs | 197 +++++++++++-- .../Runtime/EventPumpReplayGapTests.cs | 260 ++++++++++++++++++ .../Runtime/MxValueDecoderTests.cs | 141 ++++++++++ 8 files changed, 824 insertions(+), 45 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index 95571dd5..b028f14a 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -260,6 +260,16 @@ Two consequences worth knowing: - **A device that accepts a connection but serves no data will cycle**, reconnecting once per threshold. That is the correct reading of a broken device, it is rate-limited, and it is visible in the health snapshot's error count — but do not mistake it for a bug when reading logs. +- ⚠️ **"Reaches a non-Healthy state" is an obligation on each driver, and a driver can fail to meet it + without failing anything.** Galaxy's entry in the survey above is "via its reconnect supervisor" — which + holds only for the paths that actually *reach* the supervisor. Its event stream could also end the way a + closed or evicted gateway session ends it: the gRPC call completes **without an error**, so nothing + throws, the supervisor was never told, health stayed `Healthy` — and a `Healthy` driver can never satisfy + conjunct 1, so the watchdog could not fire either. The driver went permanently deaf with nothing + reporting it anywhere (fixed in `EventPump`; see [Galaxy.md](drivers/Galaxy.md) §Operational Notes). + The general shape is worth carrying to the next driver: this watchdog only covers failures a driver + **notices**, so a silent-success termination path — a stream that ends cleanly, a poll loop that exits + without throwing — bypasses it entirely. **The reconnect retry now tears down before re-initialising.** `Reconnecting`'s `RetryConnect` calls `ShutdownAsync` before `InitializeAsync` — the sequence every driver's own `ReinitializeAsync` already diff --git a/docs/drivers/Galaxy.md b/docs/drivers/Galaxy.md index 4807d4fb..4168d57e 100644 --- a/docs/drivers/Galaxy.md +++ b/docs/drivers/Galaxy.md @@ -109,3 +109,7 @@ The session-less alarm feed (`GatewayGalaxyAlarmFeed`) and alarm acknowledger (` - **MXAccess `ClientName` collisions**: two OtOpcUa instances sharing a `ClientName` cause the older Wonderware session to lose subscription state. Redundancy pairs (decision #149) enforce uniqueness via install scripts. - **Channel saturation**: `galaxy.events.dropped > 0` indicates `EventPump` is back-pressured. Raise `EventPumpChannelCapacity` or investigate downstream slowness in the server-side fan-out. - **Connectivity surface**: per-platform probe state is exposed through `IHostConnectivityProbe` and aggregated by the server's connectivity bus — there is no driver-private dashboard surface anymore. The Admin UI's Host Status panel is the consumer. +- **A stream the gateway ends cleanly is treated as a fault.** A closed or evicted gateway session completes the `StreamEvents` RPC *without an error*, so nothing throws. `EventPump` reports that to the `ReconnectSupervisor` exactly as it reports a transport fault. Before this it exited silently: the supervisor was never told, `GetHealth()` kept reporting `Healthy`, and because the [comms-loss watchdog](../DriverLifecycle.md) requires a **non-Healthy** state to trip, it never fired either — the driver went permanently deaf with no signal anywhere. Nothing else in the system can observe this, so do not remove the notification on the grounds that "a clean end isn't an error". +- **Replay gaps** (`galaxy.events.replay_gaps`): the gateway signals that events were evicted from its replay ring before we could read them. The driver responds by re-issuing `SubscribeBulk` for every tracked subscription — MXAccess delivers an initial value per item on subscribe, so re-subscribing *is* the re-snapshot the contract asks for. It does **not** restart the event pump (the stream carrying the sentinel is healthy) and it honours `Reconnect:ReplayOnSessionLost=false` by logging and standing down, on the grounds that an operator who delegated recovery to the gateway did not ask for a re-subscribe on a different trigger. + ⚠️ This cannot currently fire: the sentinel is only emitted on a stream opened with `after_worker_sequence > 0`, and the driver opens with `0`, which the gateway treats as "fresh stream, not a resume". Enabling resume is a bigger change than it looks — the sequence is only meaningful **within one gateway session**, and the reconnect path recreates the session, so a remembered sequence replayed against a fresh session would set the server's live watermark *above* every event it is about to send, silencing the stream rather than recovering it. +- **Undecodable values** (`galaxy.events.undecodable`): a value kind this driver has no CLR representation for is published as a **null under the event's own quality**, not as Bad. The status is deliberately left alone — nothing can reach this branch today, so the counter exists to produce evidence before anyone changes what an OPC UA client sees. diff --git a/docs/v2/Galaxy.Performance.md b/docs/v2/Galaxy.Performance.md index b54f1244..0e046cfa 100644 --- a/docs/v2/Galaxy.Performance.md +++ b/docs/v2/Galaxy.Performance.md @@ -38,14 +38,35 @@ the metrics surface instead. | `galaxy.events.received` | `{event}` | MxEvents read from the gateway StreamEvents stream | | `galaxy.events.dispatched` | `{event}` | MxEvents that made it through the bounded channel into `OnDataChange` | | `galaxy.events.dropped` | `{event}` | MxEvents discarded because the bounded channel was full (newest-dropped) | +| `galaxy.events.replay_gaps` | `{gap}` | Replay-gap sentinels received — the gateway evicted events we can never see | +| `galaxy.events.undecodable` | `{event}` | Data-change events carrying a value kind this driver cannot represent | -The invariant is `received = dispatched + dropped + (in-flight in the -channel)`. Watch the dropped counter — it is the leading indicator of +The invariant is `received = dispatched + dropped + replay_gaps + +(in-flight in the channel)`. A replay-gap sentinel is counted as +received but never enters the channel — it is a control signal with no +item payload, handled on the read loop precisely so channel pressure +can never drop it. + +Watch the dropped counter — it is the leading indicator of listener back-pressure. A non-zero dropped rate means a downstream consumer (DriverNodeManager → UA notification queue → client) is slower than the gw event stream; investigate that consumer before raising `EventPump` channel capacity. +`galaxy.events.replay_gaps` should be flat at zero, for a structural +reason rather than a healthy one: the gateway only emits the sentinel on +a stream opened with `after_worker_sequence > 0`, and the driver opens +with `0` ("fresh stream, not a resume"). A non-zero value therefore means +the resume path was enabled — each gap costs a full re-subscribe of every +tag on the driver, so a climbing counter is a sizing problem, not just an +observation. + +`galaxy.events.undecodable` should likewise be flat at zero: every +`MxValue` kind the contract declares decodes today. Non-zero means the +gateway shipped a new wire type and those readings are being published as +**nulls under the event's own quality** — the status is deliberately not +downgraded, so this counter is the only signal that it is happening. + ### Bounded channel design The pump runs two background tasks: diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index b3d8d537..48fda20f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -101,6 +101,13 @@ public sealed class GalaxyDriver private EventPump? _eventPump; private readonly Lock _pumpLock = new(); + /// + /// Re-entry gate for the replay-gap re-snapshot (0 = idle, 1 = running). Interlocked + /// rather than a lock because the setter runs on the event pump's read loop and must + /// never block it. + /// + private int _resnapshotInFlight; + // IAlarmSource implementation. Production-side acks route through // GatewayGalaxyAlarmAcknowledger which calls the session-less // MxGatewayClient.AcknowledgeAlarmAsync RPC; alarm transitions arrive on the @@ -464,6 +471,25 @@ public sealed class GalaxyDriver // completed loop. Recreate the pump so the replayed subscriptions have a consumer. RestartEventPumpForReplay(); + var tagCount = await ResubscribeAllAsync(entries, cancellationToken).ConfigureAwait(false); + + _logger.LogInformation( + "GalaxyDriver {InstanceId} replay completed — {SubCount} subscriptions, {TagCount} tags re-subscribed", + _driverInstanceId, entries.Count, tagCount); + } + + /// + /// Re-issues SubscribeBulk for every tracked subscription and rebinds the registry + /// with the item handles the gateway returned. Shared by the reconnect replay path and + /// the replay-gap re-snapshot — the difference between them is the event pump, which + /// replay must recreate (its loop died with the stream) and a gap must NOT touch (the + /// stream carrying the sentinel is still healthy and still delivering). + /// + /// The number of tag references re-subscribed. + private async Task ResubscribeAllAsync( + IReadOnlyList<(long SubscriptionId, IReadOnlyList Bindings)> entries, + CancellationToken cancellationToken) + { var tagCount = 0; foreach (var (subscriptionId, oldBindings) in entries) { @@ -473,7 +499,7 @@ public sealed class GalaxyDriver .ToArray(); if (refs.Length == 0) continue; - var results = await _subscriber + var results = await _subscriber! .SubscribeBulkAsync(refs, _options.MxAccess.PublishingIntervalMs, cancellationToken) .ConfigureAwait(false); @@ -493,9 +519,7 @@ public sealed class GalaxyDriver tagCount += refs.Length; } - _logger.LogInformation( - "GalaxyDriver {InstanceId} replay completed — {SubCount} subscriptions, {TagCount} tags re-subscribed", - _driverInstanceId, entries.Count, tagCount); + return tagCount; } /// @@ -1065,13 +1089,89 @@ public sealed class GalaxyDriver _subscriber!, _subscriptions, _logger, channelCapacity: _options.MxAccess.EventPumpChannelCapacity, clientName: _options.MxAccess.ClientName, - onStreamFault: OnEventPumpStreamFault); + onStreamFault: OnEventPumpStreamFault, + onReplayGap: OnEventPumpReplayGap); _eventPump.OnDataChange += OnPumpDataChange; _eventPump.Start(); return _eventPump; } } + /// + /// Replay-gap callback for the . The gateway told us it evicted + /// events we can never receive, so every cached value may be stale by an unknown amount. + /// The contract's remedy is to re-snapshot, which for this driver means re-issuing + /// SubscribeBulk: MXAccess delivers an initial value per item on subscribe, so the + /// re-subscribe is the re-snapshot. + /// + /// + /// Three things this must not do, in order of how badly they would bite: + /// + /// It must not restart the event pump. A gap is not a fault — the stream + /// that delivered the sentinel is healthy — and this runs ON that stream's read loop, + /// so disposing the pump from here would deadlock on its own loop. + /// It must not block. The read loop is stopped for as long as this runs, + /// and re-subscribing thousands of tags is a round-trip per subscription, so the work + /// goes to the thread pool and the loop returns to draining the stream immediately. + /// It must not stack. Re-entry is gated, so a burst of sentinels (or one + /// arriving while a re-snapshot is still running) collapses to a single pass rather + /// than queueing a subscribe storm per sentinel. + /// + /// is honoured here even though the + /// trigger is not a session loss: an operator who turned driver-side re-subscription off + /// is delegating recovery to the gateway, and quietly re-subscribing behind that choice + /// on a different trigger would be the same surprise they opted out of. + /// + private void OnEventPumpReplayGap(ReplayGap gap) + { + if (_subscriber is null) return; + + if (!_options.Reconnect.ReplayOnSessionLost) + { + _logger.LogWarning( + "GalaxyDriver {InstanceId} replay-gap re-snapshot skipped — ReplayOnSessionLost=false. " + + "Values missed between worker_sequence {RequestedAfter} and {OldestAvailable} stay stale " + + "until the gateway or an operator re-primes them.", + _driverInstanceId, gap.RequestedAfterSequence, gap.OldestAvailableSequence); + return; + } + + if (Interlocked.Exchange(ref _resnapshotInFlight, 1) == 1) + { + _logger.LogInformation( + "GalaxyDriver {InstanceId} replay-gap re-snapshot already in flight — coalescing.", + _driverInstanceId); + return; + } + + _ = Task.Run(async () => + { + try + { + var entries = _subscriptions.SnapshotEntries(); + if (entries.Count == 0) return; + + var tagCount = await ResubscribeAllAsync(entries, CancellationToken.None).ConfigureAwait(false); + _logger.LogInformation( + "GalaxyDriver {InstanceId} replay-gap re-snapshot completed — {SubCount} subscriptions, {TagCount} tags re-subscribed", + _driverInstanceId, entries.Count, tagCount); + } + catch (Exception ex) + { + // Never surface into the pump's read loop; the stream is still healthy and + // must keep draining. A failed re-snapshot leaves stale values, which the + // next reconnect or subscribe will correct. + _logger.LogWarning(ex, + "GalaxyDriver {InstanceId} replay-gap re-snapshot failed — cached values may stay stale.", + _driverInstanceId); + } + finally + { + Interlocked.Exchange(ref _resnapshotInFlight, 0); + } + }); + } + /// /// Stream-fault callback for the . The gw StreamEvents /// stream faulted (transient gateway drop, network blip, gw restart). Forward diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index 0d7dd12e..983add93 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -47,12 +47,19 @@ internal sealed class EventPump : IAsyncDisposable private static readonly Counter EventsDropped = Meter.CreateCounter("galaxy.events.dropped", unit: "{event}", description: "MxEvents dropped because the bounded channel was full (newest-dropped)."); + private static readonly Counter ReplayGapsObserved = + Meter.CreateCounter("galaxy.events.replay_gaps", unit: "{gap}", + description: "Replay-gap sentinels received — the gateway evicted events we can never see."); + private static readonly Counter ValuesUndecodable = + Meter.CreateCounter("galaxy.events.undecodable", unit: "{event}", + description: "Data-change events carrying a value kind this driver cannot represent."); private readonly IGalaxySubscriber _subscriber; private readonly SubscriptionRegistry _registry; private readonly ILogger _logger; private readonly Func _handleFactory; private readonly Action? _onStreamFault; + private readonly Action? _onReplayGap; private readonly Channel _channel; private readonly KeyValuePair _clientTag; private readonly CancellationTokenSource _cts = new(); @@ -60,6 +67,7 @@ internal sealed class EventPump : IAsyncDisposable private Task? _loop; private Task? _dispatchLoop; private bool _disposed; + private int _undecodableWarned; /// Occurs when a data change event is received from the Galaxy subscriber. public event EventHandler? OnDataChange; @@ -72,6 +80,10 @@ internal sealed class EventPump : IAsyncDisposable /// The bounded channel capacity for buffering events. /// The client name for metric tagging; if null, uses "<unknown>". /// Optional callback invoked when the stream faults. + /// + /// Optional callback invoked when the gateway reports a replay gap. The contract + /// obliges the client to re-snapshot; see . + /// public EventPump( IGalaxySubscriber subscriber, SubscriptionRegistry registry, @@ -79,13 +91,15 @@ internal sealed class EventPump : IAsyncDisposable Func? handleFactory = null, int channelCapacity = DefaultChannelCapacity, string? clientName = null, - Action? onStreamFault = null) + Action? onStreamFault = null, + Action? onReplayGap = null) { _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber)); _registry = registry ?? throw new ArgumentNullException(nameof(registry)); _logger = logger ?? NullLogger.Instance; _handleFactory = handleFactory ?? (id => new GalaxySubscriptionHandle(id)); _onStreamFault = onStreamFault; + _onReplayGap = onReplayGap; if (channelCapacity < 1) { @@ -130,6 +144,17 @@ internal sealed class EventPump : IAsyncDisposable if (ct.IsCancellationRequested) break; EventsReceived.Add(1, _clientTag); + // The replay-gap sentinel is a control signal, not data, and is handled HERE + // rather than in the dispatch loop for one reason: everything past this point + // is subject to newest-dropped channel pressure, and a dropped "you have + // missed events" signal is the one loss that cannot be recovered from later. + // It carries no item payload, so there is nothing to fan out downstream. + if (ev.ReplayGap is { } gap) + { + HandleReplayGap(gap); + continue; + } + // Newest-dropped: TryWrite fast-paths the common case (channel has room). // When full we count the drop and continue reading the gw stream so // back-pressure doesn't propagate upstream. @@ -138,6 +163,23 @@ internal sealed class EventPump : IAsyncDisposable EventsDropped.Add(1, _clientTag); } } + + // Falling out of the loop without an exception means the SERVER completed the + // stream — a closed or evicted gateway session, not a transport failure. That is + // just as fatal to this driver as a fault (the pump is now dead and no further + // data-change events can ever arrive) but it raises nothing, so without this the + // pump exits silently, the supervisor is never told, GetHealth() keeps reporting + // Healthy, and the #527 comms-loss watchdog never trips either — it requires a + // non-Healthy state. The result is a driver that is permanently, invisibly deaf. + if (!ct.IsCancellationRequested) + { + var cause = new InvalidOperationException( + "The gateway completed the StreamEvents stream without an error. The session was " + + "closed or evicted server-side; the event pump cannot receive further events."); + _logger.LogWarning( + "Galaxy EventPump stream was completed by the gateway — notifying reconnect supervisor."); + NotifyStreamFault(cause); + } } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -151,15 +193,7 @@ internal sealed class EventPump : IAsyncDisposable // The gw StreamEvents stream faulted. Signal the reconnect supervisor so it // drives reopen → replay. Without this the stream silently dies and a // transient gateway drop permanently stops data-change notifications. - if (_onStreamFault is not null) - { - try { _onStreamFault(ex); } - catch (Exception cbEx) - { - _logger.LogWarning(cbEx, - "Galaxy EventPump stream-fault callback threw — supervisor may not have been notified."); - } - } + NotifyStreamFault(ex); } finally { @@ -168,6 +202,54 @@ internal sealed class EventPump : IAsyncDisposable } } + private void NotifyStreamFault(Exception cause) + { + if (_onStreamFault is null) return; + try { _onStreamFault(cause); } + catch (Exception cbEx) + { + _logger.LogWarning(cbEx, + "Galaxy EventPump stream-fault callback threw — supervisor may not have been notified."); + } + } + + /// + /// Handles the gateway's replay-gap sentinel: events between the sequence we asked to + /// resume after and the oldest the gateway still retains were evicted and can never be + /// delivered. The contract is explicit that a client "MUST treat a present replay_gap as + /// 'you missed events — discard local state and re-snapshot'". + /// + /// + /// ⚠️ This is unreachable in the current configuration and that is deliberate. + /// The gateway emits the sentinel only on a stream opened with + /// after_worker_sequence > 0; GatewayGalaxySubscriber passes 0, which the + /// gateway treats as "fresh stream, not a resume: plain attach, no replay, no sentinel". + /// It is implemented anyway because the alternative — the sentinel falling through the + /// family switch into the default drop, which is what happened before, since a sentinel + /// carries family = UNSPECIFIED — means silently ignoring a documented MUST if + /// resume is ever switched on. Enabling resume is a larger change than it looks: the + /// sequence is only meaningful within one gateway session, and our reconnect path + /// recreates the session, so a remembered sequence replayed against a fresh session + /// would set the server's live watermark above every event it is about to send. + /// + private void HandleReplayGap(ReplayGap gap) + { + ReplayGapsObserved.Add(1, _clientTag); + _logger.LogWarning( + "Galaxy event stream reported a replay gap: events after worker_sequence {RequestedAfter} " + + "up to {OldestAvailable} were evicted from the gateway's replay ring and cannot be " + + "recovered. Forcing a re-snapshot of every active subscription.", + gap.RequestedAfterSequence, gap.OldestAvailableSequence); + + if (_onReplayGap is null) return; + try { _onReplayGap(gap); } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Galaxy EventPump replay-gap callback threw — the re-snapshot may not have run."); + } + } + private async Task DispatchLoopAsync(CancellationToken ct) { try @@ -230,7 +312,25 @@ internal sealed class EventPump : IAsyncDisposable private DataValueSnapshot ToSnapshot(MxEvent ev) { - var value = MxValueDecoder.Decode(ev.Value); + if (!MxValueDecoder.TryDecode(ev.Value, out var value)) + { + // The status is deliberately NOT downgraded here. Publishing null-under-Good is + // dishonest, but changing what an OPC UA client sees for a live shape is not a + // change to make blind: today nothing can reach this branch (every declared + // MxValue kind decodes), so it only fires on a wire type a future gateway adds. + // Surfacing it on a counter first means the next person has evidence to act on + // rather than a guess. Warn once per pump — a broken producer would flood. + ValuesUndecodable.Add(1, _clientTag); + if (Interlocked.Exchange(ref _undecodableWarned, 1) == 0) + { + _logger.LogWarning( + "Galaxy event for item handle {ItemHandle} carried a value kind this driver cannot " + + "represent ({KindCase}); publishing null under the event's own quality. Further " + + "occurrences are counted on galaxy.events.undecodable but not logged.", + ev.ItemHandle, ev.Value?.KindCase); + } + } + var statusCode = ev.Statuses.Count > 0 ? StatusCodeMap.FromMxStatus(ev.Statuses[0], _logger) : StatusCodeMap.FromQualityByte((byte)(ev.Quality & 0xFF), _logger); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxValueDecoder.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxValueDecoder.cs index 1f861c97..28ffb278 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxValueDecoder.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxValueDecoder.cs @@ -11,47 +11,190 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; /// back to the raw_value bytes so a forward-compatible MXAccess deployment /// doesn't lose data on the wire — the consumer can opt to deserialise the bytes. /// +/// +/// ⚠️ Every method here must be total — it must never throw. The only caller is +/// 's dispatch loop, and an exception escaping into that loop +/// terminates it permanently: the pump keeps reading the gateway stream while nothing +/// is ever dispatched again. So a malformed payload returns false from +/// rather than raising, and every bound/shape check below is a +/// guard rather than an assertion. +/// internal static class MxValueDecoder { /// Decodes a gateway MxValue into a boxed CLR object. /// The MxValue to decode, or null. - /// The decoded boxed CLR value, or null when is null or carries a null MxValue. + /// + /// The decoded boxed CLR value, or null when is null, carries a + /// null MxValue, or cannot be represented. Prefer when the caller + /// needs to tell a genuine null apart from an unrepresentable value. + /// public static object? Decode(MxValue? value) { - if (value is null) return null; - if (value.IsNull) return null; + TryDecode(value, out var decoded); + return decoded; + } - return value.KindCase switch + /// + /// Decodes a gateway MxValue, reporting whether it was representable at all. + /// + /// The MxValue to decode, or null. + /// + /// The decoded boxed CLR value. Null both for a genuine null and for a value this driver + /// cannot represent — alone cannot distinguish the two. + /// + /// + /// true when the value was decoded, including the genuine-null cases (a null + /// reference or ). false when the payload carries a + /// kind this driver has no CLR representation for — a null published under a + /// Good quality would otherwise be indistinguishable from a real null, which is + /// how a new wire type would go silently missing. + /// + public static bool TryDecode(MxValue? value, out object? decoded) + { + decoded = null; + if (value is null) return true; + if (value.IsNull) return true; + + switch (value.KindCase) { - MxValue.KindOneofCase.BoolValue => value.BoolValue, - MxValue.KindOneofCase.Int32Value => value.Int32Value, - MxValue.KindOneofCase.Int64Value => value.Int64Value, - MxValue.KindOneofCase.FloatValue => value.FloatValue, - MxValue.KindOneofCase.DoubleValue => value.DoubleValue, - MxValue.KindOneofCase.StringValue => value.StringValue, - MxValue.KindOneofCase.TimestampValue => DecodeTimestamp(value.TimestampValue), - MxValue.KindOneofCase.ArrayValue => DecodeArray(value.ArrayValue), - MxValue.KindOneofCase.RawValue => value.RawValue.ToByteArray(), - _ => null, - }; + case MxValue.KindOneofCase.BoolValue: decoded = value.BoolValue; return true; + case MxValue.KindOneofCase.Int32Value: decoded = value.Int32Value; return true; + case MxValue.KindOneofCase.Int64Value: decoded = value.Int64Value; return true; + case MxValue.KindOneofCase.FloatValue: decoded = value.FloatValue; return true; + case MxValue.KindOneofCase.DoubleValue: decoded = value.DoubleValue; return true; + case MxValue.KindOneofCase.StringValue: decoded = value.StringValue; return true; + case MxValue.KindOneofCase.TimestampValue: decoded = DecodeTimestamp(value.TimestampValue); return true; + case MxValue.KindOneofCase.RawValue: decoded = value.RawValue.ToByteArray(); return true; + case MxValue.KindOneofCase.ArrayValue: return TryDecodeArray(value.ArrayValue, out decoded); + case MxValue.KindOneofCase.SparseArrayValue: return TryDecodeSparseArray(value.SparseArrayValue, out decoded); + default: return false; + } } private static DateTime? DecodeTimestamp(Timestamp? ts) => ts?.ToDateTime(); - private static object? DecodeArray(MxArray? array) + private static bool TryDecodeArray(MxArray? array, out object? decoded) { - if (array is null) return null; + decoded = null; + if (array is null) return false; - return array.ValuesCase switch + switch (array.ValuesCase) { - MxArray.ValuesOneofCase.BoolValues => array.BoolValues.Values.ToArray(), - MxArray.ValuesOneofCase.Int32Values => array.Int32Values.Values.ToArray(), - MxArray.ValuesOneofCase.Int64Values => array.Int64Values.Values.ToArray(), - MxArray.ValuesOneofCase.FloatValues => array.FloatValues.Values.ToArray(), - MxArray.ValuesOneofCase.DoubleValues => array.DoubleValues.Values.ToArray(), - MxArray.ValuesOneofCase.StringValues => array.StringValues.Values.ToArray(), - MxArray.ValuesOneofCase.TimestampValues => array.TimestampValues.Values.Select(t => t.ToDateTime()).ToArray(), - _ => null, - }; + case MxArray.ValuesOneofCase.BoolValues: decoded = array.BoolValues.Values.ToArray(); return true; + case MxArray.ValuesOneofCase.Int32Values: decoded = array.Int32Values.Values.ToArray(); return true; + case MxArray.ValuesOneofCase.Int64Values: decoded = array.Int64Values.Values.ToArray(); return true; + case MxArray.ValuesOneofCase.FloatValues: decoded = array.FloatValues.Values.ToArray(); return true; + case MxArray.ValuesOneofCase.DoubleValues: decoded = array.DoubleValues.Values.ToArray(); return true; + case MxArray.ValuesOneofCase.StringValues: decoded = array.StringValues.Values.ToArray(); return true; + case MxArray.ValuesOneofCase.TimestampValues: + decoded = array.TimestampValues.Values.Select(t => t.ToDateTime()).ToArray(); + return true; + default: return false; + } } + + /// + /// Expands a sparse array into the dense CLR array it denotes: every index the payload + /// names carries its value, and every index it does not carries the element type's + /// default. + /// + /// + /// + /// ⚠️ Per the gateway contract this cannot arrive inbound. MxSparseArray is + /// documented write-only — the gateway expands it into a full MxArray before + /// forwarding, and "the worker never receives or produces it". This branch therefore + /// exists as contract defence, not as an expected path: if a future gateway ever emits + /// one, the alternative is a silent null published under the event's own (likely Good) + /// quality, which loses the reading with nothing to show for it. + /// + /// + /// The fill semantics deliberately mirror the gateway's own SparseArrayExpander + /// rather than inventing a reading: unmentioned indices are reset to the element + /// default, not preserved (the gateway cannot read-modify-write without racing the + /// provider), Time defaults to the Unix epoch because the expander fills + /// Timestamp{0,0}, and Integer widens to long[] if any element + /// carries an int64 — matching the expander's own Int32/Int64 sub-array choice. + /// + /// + /// Malformed payloads the gateway would have rejected outright (zero/oversized length, + /// out-of-range index) are tolerated here rather than thrown, because this side is a + /// consumer: an out-of-range index is skipped, a duplicate index is last-wins. + /// + /// + private static bool TryDecodeSparseArray(MxSparseArray? sparse, out object? decoded) + { + decoded = null; + if (sparse is null) return false; + if (sparse.TotalLength == 0 || sparse.TotalLength > (uint)Array.MaxLength) return false; + + var length = (int)sparse.TotalLength; + var elements = sparse.Elements; + + switch (sparse.ElementDataType) + { + case MxDataType.Boolean: + return Fill(new bool[length], elements, v => v.BoolValue, out decoded); + + // The expander picks its Integer sub-array width from the element kinds, since a + // sparse payload carries no CLR array to infer it from. Mirror that exactly, or a + // round-trip through the gateway would change the published CLR element type. + case MxDataType.Integer when UsesInt64(elements): + return Fill(new long[length], elements, ReadInt64, out decoded); + case MxDataType.Integer: + return Fill(new int[length], elements, v => v.Int32Value, out decoded); + + case MxDataType.Float: + return Fill(new float[length], elements, v => v.FloatValue, out decoded); + case MxDataType.Double: + return Fill(new double[length], elements, v => v.DoubleValue, out decoded); + + case MxDataType.String: + var strings = new string[length]; + Array.Fill(strings, string.Empty); + return Fill(strings, elements, v => v.StringValue ?? string.Empty, out decoded); + + case MxDataType.Time: + var times = new DateTime[length]; + Array.Fill(times, DateTime.UnixEpoch); + return Fill(times, elements, v => v.TimestampValue?.ToDateTime() ?? DateTime.UnixEpoch, out decoded); + + default: + // Structs, enums, references and the rest are not scalar element types the + // expander supports either — there is nothing meaningful to build. + return false; + } + } + + private static bool Fill( + T[] target, + IReadOnlyList elements, + Func read, + out object? decoded) + { + foreach (var element in elements) + { + if (element.Index >= (uint)target.Length) continue; // out of range — skip, never throw + if (element.Value is null) continue; // leave the default in place + target[(int)element.Index] = read(element.Value); + } + + decoded = target; + return true; + } + + private static bool UsesInt64(IReadOnlyList elements) + { + foreach (var element in elements) + { + if (element.Value?.KindCase == MxValue.KindOneofCase.Int64Value) return true; + } + return false; + } + + private static long ReadInt64(MxValue value) => value.KindCase switch + { + MxValue.KindOneofCase.Int64Value => value.Int64Value, + MxValue.KindOneofCase.Int32Value => value.Int32Value, + _ => 0L, + }; } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs new file mode 100644 index 00000000..a6f74c2c --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs @@ -0,0 +1,260 @@ +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// Coverage for the two signals that carry no item payload and so +/// have no natural home in the data-change fan-out: the gateway's replay-gap sentinel +/// (MxGateway 0.2.0) and a stream the gateway completes without an error. +/// +/// +/// +/// ⚠️ The replay-gap sentinel is unreachable in the shipped configuration. +/// GatewayGalaxySubscriber opens the stream with after_worker_sequence = 0, +/// which the gateway treats as "fresh stream, not a resume — plain attach, no replay, no +/// sentinel". These tests therefore pin a contract obligation rather than a live path: the +/// sentinel carries family = UNSPECIFIED, so before this work it fell straight +/// through the pump's family switch into the default drop, and enabling resume later would +/// have silently ignored a documented MUST. +/// +/// +/// The clean-completion half, by contrast, IS live: a gateway that closes or evicts a +/// session ends the RPC normally, which raises nothing at all. +/// +/// +public sealed class EventPumpReplayGapTests +{ + private const int WaitMs = 2_000; + + private static MxEvent GapSentinel(ulong requestedAfter, ulong oldestAvailable) => new() + { + // Shaped exactly as the gateway emits it: no family, no body, no per-item fields. + Family = MxEventFamily.Unspecified, + ReplayGap = new ReplayGap + { + RequestedAfterSequence = requestedAfter, + OldestAvailableSequence = oldestAvailable, + }, + }; + + /// Verifies the sentinel reaches the callback carrying both sequence bounds. + [Fact] + public async Task A_replay_gap_sentinel_invokes_the_callback_with_both_sequences() + { + var subscriber = new ControllableSubscriber(); + var observed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var pump = new EventPump( + subscriber, new SubscriptionRegistry(), channelCapacity: 8, clientName: "GapTest", + onReplayGap: gap => observed.TrySetResult(gap)); + pump.Start(); + + await subscriber.EmitAsync(GapSentinel(requestedAfter: 100, oldestAvailable: 250)); + + (await Task.WhenAny(observed.Task, Task.Delay(WaitMs))).ShouldBe(observed.Task, + "a replay-gap sentinel must reach the driver so it can re-snapshot"); + var gap = await observed.Task; + gap.RequestedAfterSequence.ShouldBe(100UL); + gap.OldestAvailableSequence.ShouldBe(250UL); + } + + /// + /// Verifies the sentinel is consumed as a control signal and leaves the stream working. + /// + /// + /// The "not dispatched" half is cheap insurance rather than the point: a sentinel carries + /// item handle 0, and never binds a non-positive handle + /// (that is its marker for a failed gateway subscribe), so it could not have reached a + /// subscriber even before it was intercepted. What this actually pins is that consuming + /// the sentinel off the read loop does not disturb the events around it — the load-bearing + /// assertion is the positive control that follows it. + /// + [Fact] + public async Task A_replay_gap_sentinel_is_not_dispatched_as_a_data_change() + { + var subscriber = new ControllableSubscriber(); + var registry = new SubscriptionRegistry(); + registry.Register(1, [new TagBinding("Tank.Level", ItemHandle: 7)]); + + var dispatched = 0; + var gapSeen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var realValue = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var pump = new EventPump( + subscriber, registry, channelCapacity: 8, clientName: "GapTest", + onReplayGap: _ => gapSeen.TrySetResult()); + pump.OnDataChange += (_, args) => { Interlocked.Increment(ref dispatched); realValue.TrySetResult(args); }; + pump.Start(); + + await subscriber.EmitAsync(GapSentinel(requestedAfter: 1, oldestAvailable: 2)); + (await Task.WhenAny(gapSeen.Task, Task.Delay(WaitMs))).ShouldBe(gapSeen.Task); + + // A positive control: follow the sentinel with a real event on the same handle. Without + // it, "nothing was dispatched" would also pass on a pump that dispatches nothing at all. + await subscriber.EmitAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = 7, + Value = new MxValue { DoubleValue = 12.5 }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }); + + (await Task.WhenAny(realValue.Task, Task.Delay(WaitMs))).ShouldBe(realValue.Task, + "the pump must keep dispatching real events after a sentinel"); + (await realValue.Task).Snapshot.Value.ShouldBe(12.5); + Volatile.Read(ref dispatched).ShouldBe(1, "only the real event may be dispatched, never the sentinel"); + } + + /// + /// Verifies the sentinel survives a full event channel. This is the reason it is handled + /// on the read loop rather than in the dispatch loop: everything that goes through the + /// channel is subject to newest-dropped pressure, and "you missed events" is the one + /// signal whose loss cannot be recovered from later — the pump would carry on serving + /// stale values with no indication anything was wrong. + /// + [Fact] + public async Task A_replay_gap_sentinel_survives_a_full_event_channel() + { + var subscriber = new ControllableSubscriber(); + var registry = new SubscriptionRegistry(); + registry.Register(1, [new TagBinding("Tank.Level", ItemHandle: 7)]); + + using var blockDispatch = new ManualResetEventSlim(false); + var observed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var pump = new EventPump( + subscriber, registry, channelCapacity: 1, clientName: "GapTest", + onReplayGap: gap => observed.TrySetResult(gap)); + // Wedge the dispatch loop so the bounded channel fills and stays full. + pump.OnDataChange += (_, _) => blockDispatch.Wait(TimeSpan.FromSeconds(10)); + pump.Start(); + + for (var i = 0; i < 20; i++) + { + await subscriber.EmitAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = 7, + Value = new MxValue { DoubleValue = i }, + Quality = 192, + }); + } + + await subscriber.EmitAsync(GapSentinel(requestedAfter: 5, oldestAvailable: 900)); + + (await Task.WhenAny(observed.Task, Task.Delay(WaitMs))).ShouldBe(observed.Task, + "the sentinel must not be droppable by channel back-pressure"); + (await observed.Task).OldestAvailableSequence.ShouldBe(900UL); + + blockDispatch.Set(); + } + + /// + /// Verifies a throwing gap callback cannot take the stream down with it. The callback + /// runs on the read loop, so an escaping exception would land in RunAsync's catch and be + /// reported as a transport fault — tearing down a stream that is perfectly healthy. + /// + [Fact] + public async Task A_throwing_replay_gap_callback_does_not_fault_the_stream() + { + var subscriber = new ControllableSubscriber(); + var registry = new SubscriptionRegistry(); + registry.Register(1, [new TagBinding("Tank.Level", ItemHandle: 7)]); + + var faulted = false; + var resumed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var pump = new EventPump( + subscriber, registry, channelCapacity: 8, clientName: "GapTest", + onStreamFault: _ => faulted = true, + onReplayGap: _ => throw new InvalidOperationException("re-snapshot could not be scheduled")); + pump.OnDataChange += (_, args) => resumed.TrySetResult(args); + pump.Start(); + + await subscriber.EmitAsync(GapSentinel(requestedAfter: 1, oldestAvailable: 2)); + await subscriber.EmitAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = 7, + Value = new MxValue { DoubleValue = 99.0 }, + Quality = 192, + }); + + (await Task.WhenAny(resumed.Task, Task.Delay(WaitMs))).ShouldBe(resumed.Task, + "a failed re-snapshot must not stop the stream that is still delivering"); + faulted.ShouldBeFalse("a throwing gap callback is not a transport fault"); + } + + /// + /// Verifies a stream the GATEWAY completes is reported as a fault. A closed or evicted + /// session ends the RPC normally, so nothing throws: before this, the pump exited + /// silently, the supervisor was never told, and GetHealth() kept reporting + /// Healthy — which also means the #527 comms-loss watchdog never tripped, since it + /// requires a non-Healthy state. The driver went permanently deaf with no signal at all. + /// + /// + /// The falsification control for this lives in + /// : our + /// own disposal also ends the stream, and must NOT be reported. + /// + [Fact] + public async Task A_stream_the_gateway_completes_without_an_error_is_reported_as_a_fault() + { + var subscriber = new ControllableSubscriber(); + var faultObserved = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var pump = new EventPump( + subscriber, new SubscriptionRegistry(), channelCapacity: 4, clientName: "GapTest", + onStreamFault: ex => faultObserved.TrySetResult(ex)); + pump.Start(); + + subscriber.CompleteStream(); + + (await Task.WhenAny(faultObserved.Task, Task.Delay(WaitMs))).ShouldBe(faultObserved.Task, + "a server-completed stream is as fatal as a faulted one and must reach the supervisor"); + (await faultObserved.Task).Message.ShouldContain("closed or evicted"); + } + + /// + /// fake whose stream can be fed arbitrary events and + /// completed cleanly, so both control-signal paths can be driven directly. + /// + private sealed class ControllableSubscriber : IGalaxySubscriber + { + private readonly Channel _stream = + Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true }); + + /// Subscribes to multiple tags (test stub). + /// The tag references to subscribe to. + /// The buffered update interval in milliseconds. + /// The cancellation token. + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>([]); + + /// Unsubscribes from multiple tags (test stub). + /// The item handles to unsubscribe from. + /// The cancellation token. + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) + => Task.CompletedTask; + + /// Streams events asynchronously (test stub). + /// The cancellation token. + public IAsyncEnumerable StreamEventsAsync(CancellationToken cancellationToken) + => _stream.Reader.ReadAllAsync(cancellationToken); + + /// Pushes one event onto the stream. + /// The event to emit. + public ValueTask EmitAsync(MxEvent ev) => _stream.Writer.WriteAsync(ev); + + /// Ends the stream the way a gateway closing the session does — with no error. + public void CompleteStream() => _stream.Writer.TryComplete(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/MxValueDecoderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/MxValueDecoderTests.cs index 538fc8cd..b98b63d2 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/MxValueDecoderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/MxValueDecoderTests.cs @@ -117,4 +117,145 @@ public sealed class MxValueDecoderTests var decoded = (byte[])MxValueDecoder.Decode(v)!; decoded.ShouldBe(bytes); } + + // --------------------------------------------------------------------------------- + // Sparse arrays (MxGateway 0.2.0). + // + // The gateway contract makes these write-only — it expands a sparse write into a full + // MxArray and "the worker never receives or produces it" — so inbound is a contract + // violation rather than an expected path. These pin the behaviour if one ever arrives: + // the correct dense array, never a silent null and never a throw. The fill semantics + // mirror the gateway's own SparseArrayExpander, so a value that round-trips through a + // write and back reads as the same CLR array. + // --------------------------------------------------------------------------------- + + private static MxValue Sparse(MxDataType elementType, uint totalLength, params (uint Index, MxValue Value)[] elements) + { + var sparse = new MxSparseArray { ElementDataType = elementType, TotalLength = totalLength }; + foreach (var (index, value) in elements) + { + sparse.Elements.Add(new MxSparseElement { Index = index, Value = value }); + } + return new MxValue { SparseArrayValue = sparse }; + } + + /// Verifies unmentioned indices are reset to the element default, not dropped. + [Fact] + public void Decode_SparseInt32Array_FillsUnmentionedIndicesWithTheDefault() + { + var v = Sparse(MxDataType.Integer, 5, + (1, new MxValue { Int32Value = 42 }), + (4, new MxValue { Int32Value = -7 })); + + MxValueDecoder.Decode(v).ShouldBe(new[] { 0, 42, 0, 0, -7 }); + } + + /// + /// Verifies an Integer sparse array widens to long[] when any element carries an int64. + /// The gateway picks its Int32/Int64 sub-array the same way — a sparse payload has no CLR + /// array to infer the width from — so decoding narrow here would change the published + /// element type across a write round-trip. + /// + [Fact] + public void Decode_SparseIntegerArray_WidensToInt64WhenAnyElementIs64Bit() + { + var v = Sparse(MxDataType.Integer, 3, + (0, new MxValue { Int32Value = 1 }), + (2, new MxValue { Int64Value = 8_000_000_000L })); + + MxValueDecoder.Decode(v).ShouldBe(new[] { 1L, 0L, 8_000_000_000L }); + } + + /// Verifies bool / float / double / string sparse arrays fill with their own defaults. + [Fact] + public void Decode_SparseArrays_UseThePerTypeDefaultForUnmentionedIndices() + { + MxValueDecoder.Decode(Sparse(MxDataType.Boolean, 3, (1, new MxValue { BoolValue = true }))) + .ShouldBe(new[] { false, true, false }); + + MxValueDecoder.Decode(Sparse(MxDataType.Float, 2, (0, new MxValue { FloatValue = 1.5f }))) + .ShouldBe(new[] { 1.5f, 0f }); + + MxValueDecoder.Decode(Sparse(MxDataType.Double, 2, (1, new MxValue { DoubleValue = 2.5d }))) + .ShouldBe(new[] { 0d, 2.5d }); + + // string defaults to empty, NOT null — matching the expander, so a consumer never + // has to null-check individual elements of an array it was handed whole. + MxValueDecoder.Decode(Sparse(MxDataType.String, 3, (2, new MxValue { StringValue = "z" }))) + .ShouldBe(new[] { string.Empty, string.Empty, "z" }); + } + + /// + /// Verifies a Time sparse array defaults to the Unix epoch — the expander fills + /// Timestamp{0,0}, and default(DateTime) would decode a different instant. + /// + [Fact] + public void Decode_SparseTimeArray_DefaultsToTheUnixEpochNotDefaultDateTime() + { + var when = new DateTime(2026, 8, 12, 3, 4, 5, DateTimeKind.Utc); + var v = Sparse(MxDataType.Time, 2, (1, new MxValue { TimestampValue = Timestamp.FromDateTime(when) })); + + var decoded = (DateTime[])MxValueDecoder.Decode(v)!; + decoded[0].ShouldBe(DateTime.UnixEpoch); + decoded[0].ShouldNotBe(default); + decoded[1].ShouldBe(when); + } + + /// + /// Verifies a malformed sparse payload degrades instead of throwing. The decoder runs + /// inside the pump's dispatch loop, where an escaping exception kills dispatch for the + /// life of the pump — so an out-of-range index must be skipped, not raised. + /// + [Fact] + public void Decode_SparseArray_WithAnOutOfRangeIndex_SkipsItRatherThanThrowing() + { + var v = Sparse(MxDataType.Integer, 2, + (0, new MxValue { Int32Value = 5 }), + (99, new MxValue { Int32Value = 6 })); + + object? decoded = null; + Should.NotThrow(() => decoded = MxValueDecoder.Decode(v)); + decoded.ShouldBe(new[] { 5, 0 }); + } + + /// Verifies an unsupported element type and a zero length report as unrepresentable. + [Fact] + public void TryDecode_SparseArray_WithAnUnsupportedShape_ReportsUnrepresentable() + { + // A struct element type is not a scalar the expander supports either — there is no + // array to build, so this must NOT masquerade as a successfully decoded null. + MxValueDecoder.TryDecode(Sparse(MxDataType.QualifiedStruct, 2), out var structDecoded).ShouldBeFalse(); + structDecoded.ShouldBeNull(); + + MxValueDecoder.TryDecode(Sparse(MxDataType.Integer, 0), out _).ShouldBeFalse(); + } + + // --------------------------------------------------------------------------------- + // Representable-vs-null. Decode() alone collapses both to null, which is how a value + // kind this driver cannot represent would be published as a good null and lost. + // --------------------------------------------------------------------------------- + + /// Verifies a genuine null decodes successfully rather than reporting failure. + [Fact] + public void TryDecode_DistinguishesAGenuineNullFromAnUnrepresentableValue() + { + MxValueDecoder.TryDecode(null, out var fromNullRef).ShouldBeTrue(); + fromNullRef.ShouldBeNull(); + + MxValueDecoder.TryDecode(new MxValue { IsNull = true }, out var fromIsNull).ShouldBeTrue(); + fromIsNull.ShouldBeNull(); + + // An MxValue with no kind set at all is not a null value — it is a payload this + // driver cannot interpret, and the two must not report identically. + MxValueDecoder.TryDecode(new MxValue(), out var fromEmpty).ShouldBeFalse(); + fromEmpty.ShouldBeNull(); + } + + /// Verifies an array whose element oneof is unset reports as unrepresentable. + [Fact] + public void TryDecode_ArrayWithNoElementsOneofSet_ReportsUnrepresentable() + { + MxValueDecoder.TryDecode(new MxValue { ArrayValue = new MxArray() }, out var decoded).ShouldBeFalse(); + decoded.ShouldBeNull(); + } }