feat(historian): outbox format v2 carries typed values + status; legacy entries still read

Serialize always writes v2 ("HV2\0" | ver 2 | id | tag | kind | payload | status u32 | ts). TryDeserialize reads v2 by magic, falls back to the legacy exact-length layout (also when the magic matches but v2 validation fails), else returns false - never throws. FasterLogHistorizationOutbox: RecoverState and PeekBatchAsync both use TryDeserialize. An unreadable record is indexed under a synthetic id (FIFO truncation passes it), skipped by peek, counted once (UnreadableCount), and dropped by peek once it reaches the head so it can never pin the outbox. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-09 12:50:25 -04:00
parent ff7c6c531a
commit 29893ef1b4
3 changed files with 616 additions and 36 deletions
@@ -20,7 +20,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder;
 /// </remarks>
 public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
 {
-    private readonly record struct LiveEntry(Guid Id, long Start);
+    // Unreadable = the record at Start could not be decoded (and has already been counted). Such a
+    // record is indexed under a synthetic id so FIFO truncation can pass it like any acked entry.
+    private readonly record struct LiveEntry(Guid Id, long Start, bool Unreadable = false);

    private readonly ManagedLocalStorageDevice _device;
    private readonly FasterLog _log;
@@ -41,6 +43,7 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
    private readonly Dictionary<Guid, LinkedListNode<LiveEntry>> _index = new();
    private long _nextScanAddress;   // authoritative logical head; peeks scan from here
    private long _droppedCount;
+    private long _unreadableCount;

    /// <summary>
    ///     Opens (or recovers) the FasterLog-backed outbox under <paramref name="directory"/>.
@@ -94,6 +97,24 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
        }
    }

+    /// <inheritdoc />
+    /// <remarks>
+    ///     Each undecodable record is counted once — at recovery (constructor scan) or, failing that,
+    ///     the first time a peek scans it. Such records are skipped by <see cref="PeekBatchAsync"/>,
+    ///     passed by any later FIFO ack, and dropped by a peek once they reach the head, so a corrupt
+    ///     record can never pin the head.
+    /// </remarks>
+    public long UnreadableCount
+    {
+        get
+        {
+            lock (_state)
+            {
+                return _unreadableCount;
+            }
+        }
+    }
+
    /// <inheritdoc />
    public async ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct)
    {
@@ -182,21 +203,65 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
    }

    /// <inheritdoc />
-    public ValueTask<IReadOnlyList<HistorizationOutboxEntry>> PeekBatchAsync(int max, CancellationToken ct)
+    /// <remarks>
+    ///     Undecodable records are skipped (and counted once, see <see cref="UnreadableCount"/>), never
+    ///     thrown. Unreadable records at the HEAD — ahead of every readable entry — are also removed
+    ///     here (truncate + commit), because nothing else ever would: a FIFO ack only passes records
+    ///     older than the acked entry, so a corrupt record with no readable entry behind it would
+    ///     otherwise pin the head (and the outbox count) forever.
+    /// </remarks>
+    public async ValueTask<IReadOnlyList<HistorizationOutboxEntry>> PeekBatchAsync(int max, CancellationToken ct)
    {
        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(max);

        var batch = new List<HistorizationOutboxEntry>(Math.Min(max, 64));
+        long? droppedHeadTo = null;
        lock (_state)
        {
+            // _live and the scan are both in log order, so a forward cursor pairs each scanned record
+            // with its index node (by start address) in O(n) overall.
+            LinkedListNode<LiveEntry>? cursor = _live.First;
            using FasterLogScanIterator iter = _log.Scan(_nextScanAddress, _log.TailAddress, recover: false);
-            while (batch.Count < max && iter.GetNext(out byte[] bytes, out int len, out _, out _))
+            while (batch.Count < max && iter.GetNext(out byte[] bytes, out int len, out long address, out long next))
            {
-                batch.Add(HistorizationOutboxEntrySerializer.Deserialize(bytes.AsSpan(0, len)));
+                while (cursor is not null && cursor.Value.Start < address)
+                    cursor = cursor.Next;
+                LinkedListNode<LiveEntry>? node = cursor is not null && cursor.Value.Start == address ? cursor : null;
+
+                if (HistorizationOutboxEntrySerializer.TryDeserialize(bytes.AsSpan(0, len), out HistorizationOutboxEntry? entry))
+                {
+                    batch.Add(entry);
+                    continue;
+                }
+
+                // Count once: a node flagged at recovery (or by an earlier peek) is not re-counted.
+                // An unindexed record cannot be flagged, so it would count per peek — but only this
+                // process's own (readable) appends are ever unindexed, transiently.
+                if (node is null || !node.Value.Unreadable)
+                {
+                    _unreadableCount++;
+                    if (node is not null)
+                        node.Value = node.Value with { Unreadable = true };
+                }
+
+                if (batch.Count == 0 && node is not null && ReferenceEquals(node, _live.First))
+                {
+                    cursor = node.Next;
+                    _index.Remove(node.Value.Id);
+                    _live.RemoveFirst();
+                    _nextScanAddress = next;
+                    droppedHeadTo = next;
+                }
            }
        }

-        return ValueTask.FromResult<IReadOnlyList<HistorizationOutboxEntry>>(batch);
+        if (droppedHeadTo is long head)
+        {
+            _log.TruncateUntil(head);
+            await _log.CommitAsync(ct).ConfigureAwait(false);   // make the head advance durable
+        }
+
+        return batch;
    }

    /// <inheritdoc />
@@ -254,9 +319,21 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
        using FasterLogScanIterator iter = _log.Scan(_log.BeginAddress, _log.TailAddress, recover: false);
        while (iter.GetNext(out byte[] bytes, out int len, out long currentAddress, out _))
        {
-            HistorizationOutboxEntry entry = HistorizationOutboxEntrySerializer.Deserialize(bytes.AsSpan(0, len));
-            LinkedListNode<LiveEntry> node = _live.AddLast(new LiveEntry(entry.Id, currentAddress));
-            _index[entry.Id] = node;
+            // An undecodable record must not fail construction (that would take the recorder down
+            // with it): index it under a synthetic id so FIFO truncation can pass it, flag it so
+            // PeekBatchAsync skips it without re-counting, and count it once here.
+            LiveEntry live;
+            if (HistorizationOutboxEntrySerializer.TryDeserialize(bytes.AsSpan(0, len), out HistorizationOutboxEntry? entry))
+            {
+                live = new LiveEntry(entry.Id, currentAddress);
+            }
+            else
+            {
+                live = new LiveEntry(Guid.NewGuid(), currentAddress, Unreadable: true);
+                _unreadableCount++;
+            }
+
+            _index[live.Id] = _live.AddLast(live);
        }
    }

@@ -1,6 +1,8 @@
 using System.Buffers.Binary;
+using System.Diagnostics.CodeAnalysis;
 using System.Text;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;

 namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder;

@@ -11,56 +13,256 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder;
 ///     reflection at the durable boundary.
 /// </summary>
 /// <remarks>
-///     Layout (little-endian): <c>Guid(16) | tagByteLen:int32(4) | tagUtf8(n) | value:double(8) |
-///     quality:uint16(2) | timestamp:int64(8)</c>. The timestamp is <see cref="DateTime.ToBinary"/>,
-///     which round-trips <see cref="DateTimeKind"/>.
+///     <para>
+///         <b>v2 (written):</b> <c>"HV2\0"(4) | ver=2(1) | Guid(16) | tagByteLen:int32(4) | tagUtf8(n) |
+///         kind:byte(1) | payload | status:uint32(4) | timestamp:int64(8)</c>. Payload by
+///         <see cref="HistorianValueKind"/>: Empty none; Double float64; Int64/Boolean/DateTime int64
+///         (Boolean 0/1, DateTime UTC ticks); String int32 byte length + UTF-8.
+///     </para>
+///     <para>
+///         <b>Legacy (read only):</b> <c>Guid(16) | tagByteLen:int32(4) | tagUtf8(n) | value:double(8) |
+///         quality:uint16(2) | timestamp:int64(8)</c> — exactly <c>38 + tagByteLen</c> bytes. Entries
+///         sitting in an outbox at upgrade time are read as a Double whose DA quality (always 192) maps
+///         to a StatusCode through <see cref="GatewayQualityMapper"/>.
+///     </para>
+///     <para>
+///         Timestamps are <see cref="DateTime.ToBinary"/>, which round-trips <see cref="DateTimeKind"/>.
+///         A record matching neither layout is reported unreadable by <see cref="TryDeserialize"/> —
+///         never thrown — so a corrupt record cannot jam the drain.
+///     </para>
 /// </remarks>
 internal static class HistorizationOutboxEntrySerializer
 {
-    private const ushort LegacyGoodQuality = 192;
+    /// <summary>The v2 format version byte that follows the magic.</summary>
+    internal const byte FormatVersion = 2;

-    /// <summary>Serializes <paramref name="entry"/> to a fixed-layout little-endian byte array.</summary>
+    private const int MagicLength = 4;
+    private const int V2HeaderLength = MagicLength + 1;   // magic + version
+    private const int LegacyFixedLength = 16 + 4 + 8 + 2 + 8;
+
+    private static ReadOnlySpan<byte> Magic => "HV2\0"u8;
+
+    /// <summary>Serializes <paramref name="entry"/> in the v2 layout.</summary>
    /// <param name="entry">The outbox entry to serialize.</param>
-    /// <returns>The fixed-layout little-endian byte array.</returns>
+    /// <returns>The v2 little-endian byte array.</returns>
    public static byte[] Serialize(HistorizationOutboxEntry entry)
    {
        ArgumentNullException.ThrowIfNull(entry);

+        HistorianValue value = entry.Value;
+        string? text = value.Kind == HistorianValueKind.String ? value.Text ?? string.Empty : null;
+        int payloadLen = value.Kind switch
+        {
+            HistorianValueKind.Empty => 0,
+            HistorianValueKind.Double or HistorianValueKind.Int64
+                or HistorianValueKind.Boolean or HistorianValueKind.DateTime => 8,
+            HistorianValueKind.String => 4 + Encoding.UTF8.GetByteCount(text!),
+            _ => throw new ArgumentOutOfRangeException(nameof(entry), value.Kind, "Unknown HistorianValueKind."),
+        };
+
        int tagLen = Encoding.UTF8.GetByteCount(entry.Tag);
-        var buffer = new byte[16 + 4 + tagLen + 8 + 2 + 8];
+        var buffer = new byte[V2HeaderLength + 16 + 4 + tagLen + 1 + payloadLen + 4 + 8];
        Span<byte> span = buffer;

-        entry.Id.TryWriteBytes(span[..16]);
-        BinaryPrimitives.WriteInt32LittleEndian(span.Slice(16, 4), tagLen);
-        Encoding.UTF8.GetBytes(entry.Tag, span.Slice(20, tagLen));
+        Magic.CopyTo(span);
+        span[MagicLength] = FormatVersion;
+        int p = V2HeaderLength;

-        int p = 20 + tagLen;
-        BinaryPrimitives.WriteDoubleLittleEndian(span.Slice(p, 8), entry.Value.Number);
-        p += 8;
-        BinaryPrimitives.WriteUInt16LittleEndian(span.Slice(p, 2), LegacyGoodQuality);
-        p += 2;
+        entry.Id.TryWriteBytes(span.Slice(p, 16));
+        p += 16;
+        BinaryPrimitives.WriteInt32LittleEndian(span.Slice(p, 4), tagLen);
+        p += 4;
+        Encoding.UTF8.GetBytes(entry.Tag, span.Slice(p, tagLen));
+        p += tagLen;
+
+        span[p++] = (byte)value.Kind;
+        switch (value.Kind)
+        {
+            case HistorianValueKind.Double:
+                BinaryPrimitives.WriteDoubleLittleEndian(span.Slice(p, 8), value.Number);
+                p += 8;
+                break;
+            case HistorianValueKind.Boolean:
+                // Normalised to 0/1 — the reader rejects anything else as structurally invalid.
+                BinaryPrimitives.WriteInt64LittleEndian(span.Slice(p, 8), value.Integer != 0 ? 1 : 0);
+                p += 8;
+                break;
+            case HistorianValueKind.Int64:
+            case HistorianValueKind.DateTime:
+                BinaryPrimitives.WriteInt64LittleEndian(span.Slice(p, 8), value.Integer);
+                p += 8;
+                break;
+            case HistorianValueKind.String:
+                int textLen = payloadLen - 4;
+                BinaryPrimitives.WriteInt32LittleEndian(span.Slice(p, 4), textLen);
+                p += 4;
+                Encoding.UTF8.GetBytes(text!, span.Slice(p, textLen));
+                p += textLen;
+                break;
+        }
+
+        BinaryPrimitives.WriteUInt32LittleEndian(span.Slice(p, 4), entry.StatusCode);
+        p += 4;
        BinaryPrimitives.WriteInt64LittleEndian(span.Slice(p, 8), entry.TimestampUtc.ToBinary());

        return buffer;
    }

-    /// <summary>Reconstructs a <see cref="HistorizationOutboxEntry"/> from its serialized bytes.</summary>
-    /// <param name="span">The fixed-layout little-endian bytes produced by <see cref="Serialize"/>.</param>
-    /// <returns>The reconstructed outbox entry.</returns>
-    public static HistorizationOutboxEntry Deserialize(ReadOnlySpan<byte> span)
+    /// <summary>
+    ///     Decodes a persisted record: v2 when the magic matches and the v2 structure validates, else
+    ///     the legacy layout when the length is exactly <c>38 + tagByteLen</c>, else unreadable.
+    ///     Never throws.
+    /// </summary>
+    /// <param name="span">The persisted record bytes.</param>
+    /// <param name="entry">The decoded entry, or null when unreadable.</param>
+    /// <returns><c>true</c> when the record decoded under either layout.</returns>
+    public static bool TryDeserialize(ReadOnlySpan<byte> span, [NotNullWhen(true)] out HistorizationOutboxEntry? entry)
    {
-        var id = new Guid(span[..16]);
+        // A legacy record's Guid can begin with the magic bytes, so a failed v2 parse falls through to
+        // the legacy exact-length test before the record is declared unreadable.
+        if (span.Length >= V2HeaderLength && span[..MagicLength].SequenceEqual(Magic) && TryReadV2(span, out entry))
+            return true;
+
+        return TryReadLegacy(span, out entry);
+    }
+
+    /// <summary>Decodes a persisted record, throwing when it matches neither layout.</summary>
+    /// <param name="span">The persisted record bytes.</param>
+    /// <returns>The decoded entry.</returns>
+    /// <exception cref="InvalidDataException">The record matches neither the v2 nor the legacy layout.</exception>
+    public static HistorizationOutboxEntry Deserialize(ReadOnlySpan<byte> span) =>
+        TryDeserialize(span, out HistorizationOutboxEntry? entry)
+            ? entry
+            : throw new InvalidDataException("Outbox record matches neither the v2 nor the legacy layout.");
+
+    private static bool TryReadV2(ReadOnlySpan<byte> span, [NotNullWhen(true)] out HistorizationOutboxEntry? entry)
+    {
+        entry = null;
+        if (span[MagicLength] != FormatVersion)
+            return false;
+
+        var r = new Reader(span, V2HeaderLength);
+        if (!r.TryTake(16, out ReadOnlySpan<byte> idBytes)
+            || !r.TryReadLengthPrefixed(out ReadOnlySpan<byte> tagBytes)
+            || !r.TryTake(1, out ReadOnlySpan<byte> kindByte))
+            return false;
+
+        HistorianValue value;
+        switch ((HistorianValueKind)kindByte[0])
+        {
+            case HistorianValueKind.Empty:
+                value = HistorianValue.Empty;
+                break;
+            case HistorianValueKind.Double:
+                if (!r.TryTake(8, out ReadOnlySpan<byte> d))
+                    return false;
+                value = HistorianValue.FromDouble(BinaryPrimitives.ReadDoubleLittleEndian(d));
+                break;
+            case HistorianValueKind.Int64:
+                if (!r.TryTake(8, out ReadOnlySpan<byte> l))
+                    return false;
+                value = HistorianValue.FromInt64(BinaryPrimitives.ReadInt64LittleEndian(l));
+                break;
+            case HistorianValueKind.Boolean:
+                if (!r.TryTake(8, out ReadOnlySpan<byte> b))
+                    return false;
+                long bit = BinaryPrimitives.ReadInt64LittleEndian(b);
+                if (bit is not (0 or 1))
+                    return false;
+                value = HistorianValue.FromBoolean(bit == 1);
+                break;
+            case HistorianValueKind.DateTime:
+                if (!r.TryTake(8, out ReadOnlySpan<byte> t))
+                    return false;
+                long ticks = BinaryPrimitives.ReadInt64LittleEndian(t);
+                if (ticks < DateTime.MinValue.Ticks || ticks > DateTime.MaxValue.Ticks)
+                    return false;
+                value = HistorianValue.FromDateTime(new DateTime(ticks, DateTimeKind.Utc));
+                break;
+            case HistorianValueKind.String:
+                if (!r.TryReadLengthPrefixed(out ReadOnlySpan<byte> s))
+                    return false;
+                value = HistorianValue.FromString(Encoding.UTF8.GetString(s));
+                break;
+            default:
+                return false;
+        }
+
+        if (!r.TryTake(4, out ReadOnlySpan<byte> status)
+            || !r.TryTake(8, out ReadOnlySpan<byte> ts)
+            || !r.AtEnd
+            || !TryFromBinary(BinaryPrimitives.ReadInt64LittleEndian(ts), out DateTime timestamp))
+            return false;
+
+        entry = new HistorizationOutboxEntry(
+            new Guid(idBytes), Encoding.UTF8.GetString(tagBytes), value,
+            BinaryPrimitives.ReadUInt32LittleEndian(status), timestamp);
+        return true;
+    }
+
+    private static bool TryReadLegacy(ReadOnlySpan<byte> span, [NotNullWhen(true)] out HistorizationOutboxEntry? entry)
+    {
+        entry = null;
+        if (span.Length < LegacyFixedLength)
+            return false;
+
        int tagLen = BinaryPrimitives.ReadInt32LittleEndian(span.Slice(16, 4));
-        string tag = Encoding.UTF8.GetString(span.Slice(20, tagLen));
+        if (tagLen < 0 || span.Length - LegacyFixedLength != tagLen)
+            return false;

        int p = 20 + tagLen;
-        double value = BinaryPrimitives.ReadDoubleLittleEndian(span.Slice(p, 8));
-        p += 8;
-        // Legacy layout carries a DA quality that was always 192 (Good); the typed entry
-        // carries status 0 instead, so the field is skipped.
-        p += 2;
-        long timestamp = BinaryPrimitives.ReadInt64LittleEndian(span.Slice(p, 8));
+        double number = BinaryPrimitives.ReadDoubleLittleEndian(span.Slice(p, 8));
+        ushort quality = BinaryPrimitives.ReadUInt16LittleEndian(span.Slice(p + 8, 2));
+        if (quality > byte.MaxValue
+            || !TryFromBinary(BinaryPrimitives.ReadInt64LittleEndian(span.Slice(p + 10, 8)), out DateTime timestamp))
+            return false;

-        return new HistorizationOutboxEntry(id, tag, HistorianValue.FromDouble(value), 0u, DateTime.FromBinary(timestamp));
+        entry = new HistorizationOutboxEntry(
+            new Guid(span[..16]), Encoding.UTF8.GetString(span.Slice(20, tagLen)),
+            HistorianValue.FromDouble(number), GatewayQualityMapper.Map((byte)quality), timestamp);
+        return true;
+    }
+
+    private static bool TryFromBinary(long data, out DateTime value)
+    {
+        try
+        {
+            value = DateTime.FromBinary(data);
+            return true;
+        }
+        catch (ArgumentException)
+        {
+            value = default;
+            return false;
+        }
+    }
+
+    // Bounds-checked forward cursor: every read reports false instead of throwing on a short record.
+    private ref struct Reader(ReadOnlySpan<byte> span, int position)
+    {
+        private readonly ReadOnlySpan<byte> _span = span;
+        private int _pos = position;
+
+        public readonly bool AtEnd => _pos == _span.Length;
+
+        public bool TryTake(int count, out ReadOnlySpan<byte> slice)
+        {
+            if (count < 0 || count > _span.Length - _pos)
+            {
+                slice = default;
+                return false;
+            }
+
+            slice = _span.Slice(_pos, count);
+            _pos += count;
+            return true;
+        }
+
+        public bool TryReadLengthPrefixed(out ReadOnlySpan<byte> slice)
+        {
+            slice = default;
+            return TryTake(4, out ReadOnlySpan<byte> len)
+                && TryTake(BinaryPrimitives.ReadInt32LittleEndian(len), out slice);
+        }
    }
 }