Files
lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/MxValueDecoder.cs
T
Joseph Doherty 4769044f88
v2-ci / build (push) Successful in 4m39s
v2-ci / unit-tests (push) Failing after 19m35s
feat(galaxy): handle the replay-gap sentinel and sparse-array values
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.
2026-08-12 02:56:17 -04:00

201 lines
9.9 KiB
C#

using Google.Protobuf.WellKnownTypes;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
/// <summary>
/// Translates gateway-side <see cref="MxValue"/> instances into the boxed CLR objects
/// <c>DataValueSnapshot.Value</c> carries. Mirrors the seven Galaxy data types in
/// <c>DataTypeMap</c> (Boolean, Int32, Int64, Float32, Float64, String, DateTime), plus
/// the array variants exposed by <see cref="MxArray"/>. Unknown / awkward values fall
/// back to the <c>raw_value</c> bytes so a forward-compatible MXAccess deployment
/// doesn't lose data on the wire — the consumer can opt to deserialise the bytes.
/// </summary>
/// <remarks>
/// ⚠️ <b>Every method here must be total — it must never throw.</b> The only caller is
/// <see cref="EventPump"/>'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 <c>false</c> from
/// <see cref="TryDecode"/> rather than raising, and every bound/shape check below is a
/// guard rather than an assertion.
/// </remarks>
internal static class MxValueDecoder
{
/// <summary>Decodes a gateway MxValue into a boxed CLR object.</summary>
/// <param name="value">The MxValue to decode, or null.</param>
/// <returns>
/// The decoded boxed CLR value, or null when <paramref name="value"/> is null, carries a
/// null MxValue, or cannot be represented. Prefer <see cref="TryDecode"/> when the caller
/// needs to tell a genuine null apart from an unrepresentable value.
/// </returns>
public static object? Decode(MxValue? value)
{
TryDecode(value, out var decoded);
return decoded;
}
/// <summary>
/// Decodes a gateway MxValue, reporting whether it was representable at all.
/// </summary>
/// <param name="value">The MxValue to decode, or null.</param>
/// <param name="decoded">
/// The decoded boxed CLR value. Null both for a genuine null and for a value this driver
/// cannot represent — <paramref name="decoded"/> alone cannot distinguish the two.
/// </param>
/// <returns>
/// <c>true</c> when the value was decoded, including the genuine-null cases (a null
/// reference or <see cref="MxValue.IsNull"/>). <c>false</c> when the payload carries a
/// kind this driver has no CLR representation for — a null published under a
/// <c>Good</c> quality would otherwise be indistinguishable from a real null, which is
/// how a new wire type would go silently missing.
/// </returns>
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)
{
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 bool TryDecodeArray(MxArray? array, out object? decoded)
{
decoded = null;
if (array is null) return false;
switch (array.ValuesCase)
{
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;
}
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// <para>
/// ⚠️ <b>Per the gateway contract this cannot arrive inbound.</b> <c>MxSparseArray</c> is
/// documented write-only — the gateway expands it into a full <c>MxArray</c> 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.
/// </para>
/// <para>
/// The fill semantics deliberately mirror the gateway's own <c>SparseArrayExpander</c>
/// rather than inventing a reading: unmentioned indices are <b>reset to the element
/// default, not preserved</b> (the gateway cannot read-modify-write without racing the
/// provider), <c>Time</c> defaults to the Unix epoch because the expander fills
/// <c>Timestamp{0,0}</c>, and <c>Integer</c> widens to <c>long[]</c> if any element
/// carries an <c>int64</c> — matching the expander's own Int32/Int64 sub-array choice.
/// </para>
/// <para>
/// 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.
/// </para>
/// </remarks>
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>(
T[] target,
IReadOnlyList<MxSparseElement> elements,
Func<MxValue, T> 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<MxSparseElement> 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,
};
}