From e4e313cf1cf727049d254b2753a9aeb493000135 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 24 Jul 2026 21:29:08 -0400 Subject: [PATCH] feat(mqtt): SequenceTracker seq-gap + bdSeq death-tie SequenceTracker (Driver.Mqtt/Sparkplug/) holds one edge node's Sparkplug stream-continuity state -- design doc SS3.6 invariant #3. Two independent halves: the wrapping 0-255 seq counter that reports whether a message was missed, and the bdSeq session token that ties an NDEATH to the NBIRTH it belongs to. Next-expected is (last + 1) & 0xFF, so 255 -> 0 is the sequence continuing. The boundary is the whole task because it fails differently in each direction: read the wrap as a gap and every edge node is asked to rebirth once per 256 messages, so a busy node spends its life republishing births instead of data; miss a real gap and the driver serves values it knows are behind the device, at Good quality, indefinitely. Both directions are asserted, plus two full wrap cycles so an off-by-one that only misfires at one value has nowhere to hide. A gap is reported once and the tracker then resynchronizes onto the observed value -- holding the baseline at the pre-gap number would make every following message report a gap too, so one lost message would become a permanent rebirth storm, the same pathology as a wrong wrap reached by a different route. An out-of-range seq is refused, never masked. The codec hands seq over as a ulong precisely so a spec-violating publisher's value is not truncated into a plausible-looking one, and masking would finish the job it declined to do: with the baseline at 255, a seq of 256 masks to 0 -- exactly what the wrap expects next -- and the bogus message would be accepted as contiguous. Refused values do not become the baseline either, so the next genuine message is still measured against the last credible one. A birth restarts the sequence via AcceptBirth rather than Accept, so it is never itself a gap; a gap after a birth is still detected. An in-range but non-zero birth seq is adopted rather than refused -- refusing it would demand a fresh rebirth for every message from an otherwise-followable publisher. The first message on a virgin tracker cannot be a gap (nothing to measure it against) but leaves IsBirthSynchronized false, which is how Task 21 tells a mid-stream late join from a synchronized node. bdSeq exists to stop a late Last Will from killing a live node: a node's connection drops, it reconnects and publishes a fresh NBIRTH, and only then does the broker notice the old connection died and deliver its will carrying the PREVIOUS session's bdSeq. Without the compare that stale NDEATH drives every tag of a freshly-born healthy node to Bad. Unknowns fail toward stale deliberately -- only a positive mismatch of two known tokens discards a death, because acting on a death wrongly self-corrects at the next birth (invariant #4) while ignoring a real one leaves a dead node's values flowing Good forever. bdSeq is not a top-level field, so TryReadBdSeq extracts it from the metric named bdSeq and runs it through ReinterpretSigned first: read raw, a negative Int64 bdSeq becomes an enormous ulong -- a plausible-looking session token minted out of nonsense. Falsifiability verified, each mutation reverted after observing RED: (a) no-wrap expected -> 2 RED; (b) Accept always true -> 5 RED; (c) bdSeq compare always matches -> 2 RED; (d) mask out-of-range seq into range -> 2 RED. 28 tests, 437/437 for the MQTT suite. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../Sparkplug/SequenceTracker.cs | 375 +++++++++++++++ .../SequenceTrackerTests.cs | 451 ++++++++++++++++++ 2 files changed, 826 insertions(+) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SequenceTrackerTests.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs new file mode 100644 index 00000000..13c91bb3 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs @@ -0,0 +1,375 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +/// +/// Per-edge-node Sparkplug B stream-continuity state: the wrapping seq counter that says +/// whether the driver has missed a message, and the bdSeq session token that ties an +/// NDEATH to the NBIRTH it belongs to. Design doc §3.6 invariant #3. +/// +/// +/// +/// One instance per edge node — this object is the edge node's stream state. It +/// holds no key and no map. Sparkplug numbers messages per edge node (an edge node's devices +/// share its counter, so a DDATA advances the same sequence an NDATA does), and the ingest +/// state machine already keeps per-scope state for the alias table and birth cache; giving +/// this type a second internal dictionary would mean two maps whose lifetimes must be kept in +/// step by hand, and a tracker that outlived a removed edge node would go on answering +/// questions about a stream nobody is reading. +/// +/// +/// This type detects; it does not react. It never publishes, never requests a rebirth +/// and never emits a quality change — the ingest state machine reads its verdicts and decides. +/// The split matters because "was a message lost" is a fact about the wire, while "request a +/// rebirth now" is a policy gated on requestRebirthOnGap. +/// +/// +/// Thread-safe under a plain monitor. Sparkplug messages arrive on MQTTnet's shared +/// dispatcher thread while the reconnect path can call from another, and +/// every operation here is a read-modify-write of two coupled fields — the immutable-snapshot +/// discipline MqttSubscriptionManager uses fits a set that is swapped wholesale, not a +/// counter that is advanced. The critical sections are a handful of instructions each and +/// nothing inside them can block, so the simple lock this repo uses elsewhere +/// (MqttDriver._hostLock) is the right shape. +/// +/// +/// Nothing here throws, for any input. Same reasoning as +/// : this sits behind an unauthenticated firehose on the +/// dispatcher thread, so a spec-violating publisher gets a verdict, never an exception that +/// would stall delivery for every subscription on the connection. +/// +/// +public sealed class SequenceTracker +{ + /// + /// The largest legal Sparkplug seq. The counter is a byte on the wire in spirit but a + /// uint64 in the schema, so the range check lives here — see . + /// + public const ulong MaxSeq = 255; + + private readonly object _gate = new(); + + /// The last credible sequence number, meaningful only when . + private byte _lastSeq; + + /// Whether any credible sequence number has been observed since the last reset. + private bool _hasLastSeq; + + /// Whether the current baseline was established by a birth rather than adopted mid-stream. + private bool _birthSynchronized; + + /// The current session's bdSeq, or null when no birth has supplied one. + private ulong? _birthBdSeq; + + /// + /// Whether the current sequence baseline came from a birth. also after + /// a first message that was not a birth — the driver joined mid-stream and holds no + /// alias table, which is a different situation from a gap even though both call for a rebirth. + /// + public bool IsBirthSynchronized + { + get { lock (_gate) return _birthSynchronized; } + } + + /// + /// The last credible sequence number observed, or when none has been — + /// before the first message, or after . + /// + public byte? LastSeq + { + get { lock (_gate) return _hasLastSeq ? _lastSeq : null; } + } + + /// + /// The bdSeq of the most recent birth, or when no birth has + /// supplied one. This is the token matches against. + /// + public ulong? BirthBdSeq + { + get { lock (_gate) return _birthBdSeq; } + } + + /// + /// Records a birth (NBIRTH/DBIRTH): restarts the sequence at the birth's seq and adopts + /// its bdSeq as the current session token. + /// + /// + /// The birth payload's seq — , which is + /// when the message carried none. + /// + /// + /// The birth's session token, read from its bdSeq metric with + /// ; when the birth carried none. + /// + /// + /// when the birth established a usable sequence baseline; + /// when its seq was absent or out of range. + /// + /// + /// + /// A birth is never a gap. It restarts the sequence by definition, so it must not + /// be run through — doing so would flag a gap on the very message + /// that just resynchronized everything, and answer a birth with a request for another one. + /// + /// + /// An in-range seq other than the spec-required 0 is adopted, not refused. + /// Refusing it would leave the tracker unsynchronized against a publisher whose stream is + /// otherwise perfectly followable, and every message that followed would demand a fresh + /// rebirth — the storm this type exists to prevent, arrived at by being strict. + /// + /// + /// The bdSeq is recorded even when the seq was unusable. Session + /// identity and position-in-stream are independent facts; discarding the tie because the + /// sequence number was garbled would leave the next NDEATH unmatchable, which is the one + /// thing that lets a stale will kill a live node. + /// + /// + public bool AcceptBirth(ulong? seq, ulong? bdSeq = null) + { + lock (_gate) + { + _birthBdSeq = bdSeq; + + if (seq is not { } value || value > MaxSeq) + { + _hasLastSeq = false; + _birthSynchronized = false; + return false; + } + + _lastSeq = (byte)value; + _hasLastSeq = true; + _birthSynchronized = true; + return true; + } + } + + /// + /// Offers the next sequenced message's seq (NDATA/DDATA/DDEATH) and reports whether the + /// stream is still contiguous. + /// + /// + /// The payload's seq — , which is + /// when the message carried none. + /// + /// + /// when this seq is the one that follows the last; + /// when a message was missed, when the value is out of range, or when + /// it is absent. A is the caller's cue to request a rebirth. + /// + /// + /// + /// The wrap is the whole point: next-expected is (last + 1) & 0xFF, so + /// 255 → 0 is the sequence continuing, not a gap. Getting the boundary wrong fails + /// loudly in one direction and silently in the other. Read the wrap as a gap and every + /// edge node is asked to rebirth once per 256 messages, so a busy node spends its life + /// republishing births instead of publishing data. Miss a real gap and the driver serves + /// values it knows are behind the device, at Good quality, indefinitely. + /// + /// + /// A gap is reported once, then the tracker resynchronizes onto the observed value. + /// Holding the baseline at the pre-gap number would make every following message report a + /// gap too, so a single lost message would become a permanent rebirth storm. + /// + /// + /// An out-of-range seq is refused, never masked into range. The codec hands + /// this over as a precisely so a spec-violating publisher's value is + /// not silently truncated into a plausible-looking one, and masking would finish the job + /// it declined to do: with the baseline at 255, a seq of 256 masks to 0 — exactly + /// the value the wrap expects next — and the bogus message would be accepted as + /// contiguous. Refused values do not become the baseline either, so the next genuine + /// message is still measured against the last credible one. + /// + /// + /// The first message is not a gap — there is nothing to measure it against. It is + /// adopted as the baseline, and stays + /// to record that the driver joined mid-stream. + /// + /// + /// NDEATH must not be offered here. An NDEATH is the broker-published Last Will and + /// carries no seq at all; feeding it in would report a gap on a message that never + /// had a place in the sequence. It is tied to its birth by + /// instead. A DDEATH is published by the edge node + /// and is sequenced, so it does belong here. + /// + /// + public bool Accept(ulong? seq) + { + lock (_gate) + { + // Refused without becoming the baseline: see the out-of-range remarks above. + if (seq is not { } value || value > MaxSeq) + { + return false; + } + + var current = (byte)value; + + if (!_hasLastSeq) + { + _lastSeq = current; + _hasLastSeq = true; + return true; + } + + var expected = (byte)((_lastSeq + 1) & 0xFF); + + // Resynchronize even on a gap — one lost message must not become a permanent storm. + _lastSeq = current; + + return current == expected; + } + } + + /// + /// Reports whether an NDEATH belongs to the session currently believed alive — the + /// bdSeq tie of §3.6 invariant #3. + /// + /// + /// The death payload's bdSeq, read with ; + /// when it carried none. + /// + /// + /// when the death applies and the node's metrics should go stale; + /// when it is a previous session's will and must be ignored. + /// + /// + /// + /// What this prevents. An edge node's connection drops, it reconnects and publishes + /// a fresh NBIRTH — and only then does the broker notice the old connection died + /// and deliver its retained Last Will. The NDEATH arrives after the NBIRTH, carrying the + /// previous session's bdSeq. Without this comparison that stale will marks a live, + /// freshly-born node dead and drives every one of its tags to Bad quality, with no further + /// message coming to correct it until the node happens to rebirth again. + /// + /// + /// Unknowns fail toward stale, deliberately. A death arriving before any birth was + /// seen, a death carrying no bdSeq, or a birth that carried none all resolve to + /// . The two error directions are not symmetric: acting on a death + /// wrongly marks tags stale until the next birth restores them (§3.6 invariant #4, a + /// self-correcting error an operator can see), while ignoring a real death leaves a dead + /// node's last values flowing at Good quality forever. Only a positive mismatch — + /// two known, different tokens — is evidence enough to discard a death. + /// + /// + /// Pure. It answers a question and changes nothing, so the caller can ask it while + /// logging or deciding. A death that is acted on needs no reset here: the node is + /// offline, and the next birth calls which restarts the sequence + /// anyway. + /// + /// + public bool IsDeathForCurrentSession(ulong? deathBdSeq) + { + lock (_gate) + { + if (_birthBdSeq is not { } birth || deathBdSeq is not { } death) + { + return true; + } + + return birth == death; + } + } + + /// + /// Returns the tracker to its virgin state: no sequence baseline, not birth-synchronized, no + /// session token. + /// + /// + /// The reconnect seam (§3.6 invariant #5 — late join). After a dropped connection the driver + /// has missed an unknown number of messages, so the baseline it was holding is no longer + /// evidence of anything: carrying it across would let the stream look contiguous purely + /// because the edge node happened to publish the right number of messages while the driver was + /// away. + /// + public void Reset() + { + lock (_gate) + { + _lastSeq = 0; + _hasLastSeq = false; + _birthSynchronized = false; + _birthBdSeq = null; + } + } + + /// + /// Reads the bdSeq session token out of a decoded NBIRTH or NDEATH payload. + /// + /// The decoded payload; and invalid are both handled. + /// The token on success; 0 otherwise. + /// when the payload carried a usable bdSeq metric. + /// + /// + /// bdSeq is not a top-level payload field — unlike seq it rides as an + /// ordinary metric named bdSeq, which is why reading it is a helper rather than a + /// property. The name is matched exactly (ordinal): the spec fixes it, so a case variant + /// is far more likely to be a different metric than a typo of this one. + /// + /// + /// Signed values are reinterpreted before they are judged. Sparkplug carries + /// Int64 as two's complement in an unsigned field, so a negative bdSeq + /// reaches the projection as an enormous . Read raw it would become a + /// plausible-looking session token minted out of nonsense; run through + /// it is visibly negative and refused. + /// + /// + /// The value is treated as an opaque token, not as a 0–255 counter. Its only job + /// here is equality against the birth's, and a range check could never make a match more + /// or less correct — it could only discard the one piece of evidence tying a death to its + /// birth. This is the opposite call from seq, whose range is load-bearing precisely + /// because its arithmetic wraps. + /// + /// + public static bool TryReadBdSeq(SparkplugPayload? payload, out ulong bdSeq) + { + bdSeq = 0; + + if (payload is not { IsValid: true }) + { + return false; + } + + foreach (var metric in payload.Metrics) + { + if (!string.Equals(metric.Name, "bdSeq", StringComparison.Ordinal)) + { + continue; + } + + if (metric.ValueKind != SparkplugValueKind.Scalar) + { + return false; + } + + var value = metric.DataType is { } datatype + ? SparkplugCodec.ReinterpretSigned(metric.Value, datatype) + : metric.Value; + + switch (value) + { + case ulong u: + bdSeq = u; + return true; + case uint u: + bdSeq = u; + return true; + case long l when l >= 0: + bdSeq = (ulong)l; + return true; + case int i when i >= 0: + bdSeq = (ulong)i; + return true; + case short s when s >= 0: + bdSeq = (ulong)s; + return true; + case sbyte sb when sb >= 0: + bdSeq = (ulong)sb; + return true; + default: + // Negative, or a wire type no integer token can come from. + return false; + } + } + + return false; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SequenceTrackerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SequenceTrackerTests.cs new file mode 100644 index 00000000..968a56c8 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SequenceTrackerTests.cs @@ -0,0 +1,451 @@ +using Org.Eclipse.Tahu.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// Covers — design doc §3.6 invariant #3. Two halves, each with a +/// distinct failure mode the tests below pin from both directions: +/// +/// +/// seq gap detection across the 0–255 wrap. Reading 255→0 as a gap +/// rebirth-storms every edge node once per 256 messages; missing a real gap serves stale +/// data at Good quality forever. Both directions are asserted. +/// +/// +/// bdSeq birth↔death tie. A late Last-Will NDEATH from a previous session +/// arriving after the new NBIRTH must not kill the freshly-born node. +/// +/// +/// +public sealed class SequenceTrackerTests +{ + // --------------------------------------------------------------------------------------- + // seq — the wrap boundary + // --------------------------------------------------------------------------------------- + + /// + /// The plan's headline case: 255 → 0 is the sequence continuing, not a gap. A tracker that + /// reads the wrap as a gap requests a rebirth every 256 messages, and a busy edge node then + /// spends its life republishing births instead of data. + /// + [Fact] + public void Sequence_WrapAt255IsContiguous_NotAGap() + { + var s = new SequenceTracker(); + + s.Accept(254).ShouldBeTrue(); + s.Accept(255).ShouldBeTrue(); + s.Accept(0).ShouldBeTrue(); // wrap + } + + /// + /// The plan's mirror case: a skipped value is a genuine gap. Missing it means the driver + /// silently keeps serving values it knows are behind the edge node's real state. + /// + [Fact] + public void Sequence_SkippedValue_IsGap() + { + var s = new SequenceTracker(); + + s.Accept(10).ShouldBeTrue(); + s.Accept(12).ShouldBeFalse(); // gap → caller requests rebirth + } + + /// + /// Two-and-a-bit full cycles of the 0–255 range report no gap at any point. The pair of + /// single-step tests above can both pass on an off-by-one wrap that only misfires at one + /// specific value; walking every value in the range leaves nowhere for that to hide. + /// + [Fact] + public void Sequence_FullWrapCycles_NeverReportAGap() + { + var s = new SequenceTracker(); + s.AcceptBirth(0).ShouldBeTrue(); + + for (var i = 1; i <= 600; i++) + { + var seq = (ulong)(i % 256); + s.Accept(seq).ShouldBeTrue($"seq {seq} (message {i}) was reported as a gap"); + } + } + + /// + /// A gap is reported once and the tracker then resynchronizes onto the observed + /// sequence. Leaving the baseline pinned at the pre-gap value would make every subsequent + /// message report a gap too, turning one lost message into a permanent rebirth storm — the + /// same pathology as a wrong wrap boundary, reached by a different route. + /// + [Fact] + public void Sequence_GapReportedOnce_ThenResynchronizes() + { + var s = new SequenceTracker(); + + s.Accept(10).ShouldBeTrue(); + s.Accept(13).ShouldBeFalse(); // three messages lost + s.Accept(14).ShouldBeTrue(); // resynchronized onto 13 + s.Accept(15).ShouldBeTrue(); + } + + /// + /// A repeated sequence number is not contiguous. Sparkplug publishes DATA at QoS 0, so a + /// duplicate is not a broker redelivery — it is a publisher whose stream no longer matches + /// what the driver can reason about, and a rebirth is the cheap correct answer. + /// + [Fact] + public void Sequence_RepeatedValue_IsGap() + { + var s = new SequenceTracker(); + + s.Accept(10).ShouldBeTrue(); + s.Accept(10).ShouldBeFalse(); + } + + /// A sequence number going backwards is a gap, not a wrap. + [Fact] + public void Sequence_BackwardJump_IsGap() + { + var s = new SequenceTracker(); + + s.Accept(10).ShouldBeTrue(); + s.Accept(9).ShouldBeFalse(); + } + + /// + /// The first message cannot be a gap — there is no baseline to measure it against — but the + /// tracker must still report that it is not birth-synchronized, because a first + /// message that was not a birth means the driver joined mid-stream and has no alias table. + /// Conflating the two would let a late join look like a healthy synchronized node. + /// + [Fact] + public void Sequence_FirstMessage_IsNotAGap_ButIsNotBirthSynchronized() + { + var s = new SequenceTracker(); + + s.IsBirthSynchronized.ShouldBeFalse(); + s.LastSeq.ShouldBeNull(); + + s.Accept(137).ShouldBeTrue(); + + s.LastSeq.ShouldBe((byte)137); + s.IsBirthSynchronized.ShouldBeFalse("a mid-stream first message is a late join, not a birth"); + } + + /// + /// The aliasing test. The codec hands over seq as a and + /// deliberately declines to truncate it, so an out-of-range value reaches this type intact. A + /// tracker that masks it into range turns 256 into a perfectly plausible 0 — and with the + /// baseline sitting at 255, that is exactly the value the wrap expects next, so the bogus + /// message would be accepted as contiguous. + /// + [Fact] + public void Sequence_OutOfRangeSeq_IsRejected_AndDoesNotAliasOntoTheExpectedValue() + { + var s = new SequenceTracker(); + s.Accept(255).ShouldBeTrue(); + + // 256 & 0xFF == 0, and 0 is precisely what follows 255. Masking would return true here. + s.Accept(256).ShouldBeFalse(); + } + + /// + /// A rejected out-of-range value does not become the baseline either: the tracker keeps the + /// last credible sequence number, so the following genuine message is still measured against + /// something real. + /// + [Fact] + public void Sequence_OutOfRangeSeq_DoesNotBecomeTheBaseline() + { + var s = new SequenceTracker(); + s.Accept(10).ShouldBeTrue(); + + s.Accept(ulong.MaxValue).ShouldBeFalse(); + + s.LastSeq.ShouldBe((byte)10); + s.Accept(11).ShouldBeTrue("the baseline must still be the last credible seq, 10"); + } + + /// + /// A message carrying no sequence number at all is a spec violation the tracker cannot place + /// in the stream. It reports a gap (the safe direction — a rebirth resynchronizes everything) + /// and leaves the baseline alone. + /// + [Fact] + public void Sequence_AbsentSeq_IsGap_AndDoesNotAdvanceTheBaseline() + { + var s = new SequenceTracker(); + s.Accept(10).ShouldBeTrue(); + + s.Accept(null).ShouldBeFalse(); + + s.LastSeq.ShouldBe((byte)10); + s.Accept(11).ShouldBeTrue(); + } + + // --------------------------------------------------------------------------------------- + // seq — births restart the sequence + // --------------------------------------------------------------------------------------- + + /// + /// A birth restarts the sequence at 0, so it is never itself a gap however far the previous + /// session had counted. A tracker that ran the birth through the ordinary contiguity check + /// would flag a gap on the very message that just fixed everything, and request a rebirth in + /// response to a birth. + /// + [Fact] + public void Birth_RestartsTheSequence_SoABirthIsNeverAGap() + { + var s = new SequenceTracker(); + s.Accept(200).ShouldBeTrue(); + + s.AcceptBirth(0).ShouldBeTrue(); + + s.LastSeq.ShouldBe((byte)0); + s.IsBirthSynchronized.ShouldBeTrue(); + s.Accept(1).ShouldBeTrue(); + } + + /// + /// A gap detected after a birth is genuine — the birth resets the baseline, it does not + /// suppress detection. Without this the previous test could be satisfied by a tracker that + /// simply stopped checking once a birth arrived. + /// + [Fact] + public void Birth_ThenSkippedValue_IsStillAGap() + { + var s = new SequenceTracker(); + s.AcceptBirth(0).ShouldBeTrue(); + + s.Accept(1).ShouldBeTrue(); + s.Accept(3).ShouldBeFalse(); + } + + /// + /// A birth whose seq is in range but not the spec-required 0 is adopted as the + /// baseline rather than rejected. Refusing it would leave the tracker unsynchronized against + /// a publisher whose stream is otherwise perfectly followable, and every following message + /// would demand another rebirth. + /// + [Fact] + public void Birth_WithNonZeroInRangeSeq_IsAdoptedAsTheBaseline() + { + var s = new SequenceTracker(); + + s.AcceptBirth(7).ShouldBeTrue(); + + s.LastSeq.ShouldBe((byte)7); + s.IsBirthSynchronized.ShouldBeTrue(); + s.Accept(8).ShouldBeTrue(); + } + + /// + /// A birth whose seq is out of range establishes no baseline — the same + /// no-masking rule as the data path, applied where it matters most. + /// + [Fact] + public void Birth_WithOutOfRangeSeq_EstablishesNoBaseline() + { + var s = new SequenceTracker(); + + s.AcceptBirth(256).ShouldBeFalse(); + + s.LastSeq.ShouldBeNull(); + s.IsBirthSynchronized.ShouldBeFalse(); + } + + /// + /// returns the tracker to its virgin state. This is the + /// reconnect seam: after a dropped connection the driver has missed an unknown number of + /// messages, so the baseline it held is no longer evidence of anything. + /// + [Fact] + public void Reset_ClearsTheBaselineAndTheBirthSynchronizedFlag() + { + var s = new SequenceTracker(); + s.AcceptBirth(0, bdSeq: 4).ShouldBeTrue(); + s.Accept(1).ShouldBeTrue(); + + s.Reset(); + + s.LastSeq.ShouldBeNull(); + s.IsBirthSynchronized.ShouldBeFalse(); + s.BirthBdSeq.ShouldBeNull(); + s.Accept(99).ShouldBeTrue("a virgin tracker cannot detect a gap on its first message"); + } + + // --------------------------------------------------------------------------------------- + // bdSeq — the birth↔death tie + // --------------------------------------------------------------------------------------- + + /// + /// The case the tie exists for. A node's connection drops, it reconnects and publishes + /// a fresh NBIRTH — and only then does the broker notice the old connection died and deliver + /// its Last Will NDEATH, carrying the previous session's bdSeq. Without the compare, + /// that stale will marks a live, freshly-born node dead and every one of its tags goes Bad. + /// + [Fact] + public void Death_WithStaleBdSeqArrivingAfterARebirth_DoesNotApply() + { + var s = new SequenceTracker(); + s.AcceptBirth(0, bdSeq: 7); // session 7 — the one that is about to die + s.AcceptBirth(0, bdSeq: 8); // session 8 — the node is back, and alive + + s.IsDeathForCurrentSession(7).ShouldBeFalse("session 7's will must not kill session 8"); + } + + /// A death whose bdSeq matches the current birth is the real thing and applies. + [Fact] + public void Death_WithMatchingBdSeq_Applies() + { + var s = new SequenceTracker(); + s.AcceptBirth(0, bdSeq: 8); + + s.IsDeathForCurrentSession(8).ShouldBeTrue(); + } + + /// + /// A death arriving before any birth was ever seen applies. The tracker has nothing to + /// disprove it with, and the two failure directions are not symmetric: acting on a death + /// wrongly marks tags stale until the next birth restores them (§3.6 invariant #4), while + /// ignoring a real death leaves a dead node's last values flowing at Good quality forever. + /// + [Fact] + public void Death_BeforeAnyBirth_Applies() + { + var s = new SequenceTracker(); + + s.IsDeathForCurrentSession(7).ShouldBeTrue(); + } + + /// + /// A death carrying no bdSeq metric cannot be tied to anything, so it applies — same + /// fail-toward-stale reasoning as above. + /// + [Fact] + public void Death_WithNoBdSeq_Applies() + { + var s = new SequenceTracker(); + s.AcceptBirth(0, bdSeq: 8); + + s.IsDeathForCurrentSession(null).ShouldBeTrue(); + } + + /// A death applies when the birth was the one that carried no bdSeq. + [Fact] + public void Death_WhenTheBirthCarriedNoBdSeq_Applies() + { + var s = new SequenceTracker(); + s.AcceptBirth(0, bdSeq: null); + + s.IsDeathForCurrentSession(7).ShouldBeTrue(); + } + + /// + /// A birth records its bdSeq even when its seq was unusable. Session identity and + /// sequence position are independent facts; discarding the tie because the sequence number + /// was garbled would leave the next death unmatchable. + /// + [Fact] + public void Birth_WithUnusableSeq_StillRecordsItsBdSeq() + { + var s = new SequenceTracker(); + + s.AcceptBirth(256, bdSeq: 8).ShouldBeFalse(); + + s.BirthBdSeq.ShouldBe(8UL); + s.IsDeathForCurrentSession(7).ShouldBeFalse(); + } + + // --------------------------------------------------------------------------------------- + // bdSeq — reading it out of a payload + // --------------------------------------------------------------------------------------- + + /// + /// bdSeq is not a top-level payload field: it rides as a metric named bdSeq in + /// the NBIRTH and the NDEATH. + /// + [Fact] + public void TryReadBdSeq_ReadsTheBdSeqMetric() + { + var payload = PayloadWith(Metric("bdSeq", DataType.Int64, 8UL)); + + SequenceTracker.TryReadBdSeq(payload, out var bdSeq).ShouldBeTrue(); + bdSeq.ShouldBe(8UL); + } + + /// + /// Sparkplug carries Int64 as two's complement in an unsigned field, so a negative + /// bdSeq reaches the projection as an enormous . Reading it raw would + /// mint a plausible-looking session token out of nonsense; the reinterpretation is what + /// exposes it as negative so it can be refused. + /// + [Fact] + public void TryReadBdSeq_NegativeSignedValue_IsRefused_NotReadAsAHugeToken() + { + var payload = PayloadWith(Metric("bdSeq", DataType.Int64, unchecked((ulong)-1L))); + + SequenceTracker.TryReadBdSeq(payload, out var bdSeq).ShouldBeFalse(); + bdSeq.ShouldBe(0UL); + } + + /// + /// An unsigned bdSeq is read as-is — no reinterpretation applies. (The generated enum member + /// is Uint64, not UInt64: protobuf's C# codegen lowercases the interior of the + /// proto's UInt64.) + /// + [Fact] + public void TryReadBdSeq_UnsignedValue_IsReadAsIs() + { + var payload = PayloadWith(Metric("bdSeq", DataType.Uint64, 9UL)); + + SequenceTracker.TryReadBdSeq(payload, out var bdSeq).ShouldBeTrue(); + bdSeq.ShouldBe(9UL); + } + + /// A payload that carries no bdSeq metric reports absence rather than a default. + [Fact] + public void TryReadBdSeq_PayloadWithoutTheMetric_ReportsAbsence() + { + var payload = PayloadWith(Metric("Temperature", DataType.Float, 21.5f)); + + SequenceTracker.TryReadBdSeq(payload, out var bdSeq).ShouldBeFalse(); + bdSeq.ShouldBe(0UL); + } + + /// + /// An undecodable payload yields no bdSeq. It presents as metric-less exactly like a valid + /// but empty one, and only tells them apart. + /// + [Fact] + public void TryReadBdSeq_InvalidPayload_ReportsAbsence() + { + SequenceTracker.TryReadBdSeq(SparkplugPayload.Invalid, out _).ShouldBeFalse(); + } + + /// A null payload is a verdict, not an exception — this runs on the dispatcher thread. + [Fact] + public void TryReadBdSeq_NullPayload_ReportsAbsence_DoesNotThrow() + { + Should.NotThrow(() => SequenceTracker.TryReadBdSeq(null, out _).ShouldBeFalse()); + } + + /// + /// A bdSeq metric with no scalar value (explicitly null, or a kind v1 does not support) is + /// not a session token. + /// + [Fact] + public void TryReadBdSeq_NonScalarMetric_ReportsAbsence() + { + var payload = PayloadWith(new SparkplugMetric("bdSeq", null, DataType.Int64, null, SparkplugValueKind.Null, null)); + + SequenceTracker.TryReadBdSeq(payload, out _).ShouldBeFalse(); + } + + private static SparkplugPayload PayloadWith(params SparkplugMetric[] metrics) => + new(IsValid: true, Seq: 0UL, TimestampMs: 1UL, Metrics: metrics); + + private static SparkplugMetric Metric(string name, DataType dataType, object value) => + new(name, Alias: null, dataType, TimestampMs: null, SparkplugValueKind.Scalar, value); +}