diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs index a1daa0c9..1698778c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs @@ -20,7 +20,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; /// 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> _index = new(); private long _nextScanAddress; // authoritative logical head; peeks scan from here private long _droppedCount; + private long _unreadableCount; /// /// Opens (or recovers) the FasterLog-backed outbox under . @@ -94,6 +97,24 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox } } + /// + /// + /// 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 , + /// 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. + /// + public long UnreadableCount + { + get + { + lock (_state) + { + return _unreadableCount; + } + } + } + /// public async ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) { @@ -182,21 +203,65 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox } /// - public ValueTask> PeekBatchAsync(int max, CancellationToken ct) + /// + /// Undecodable records are skipped (and counted once, see ), 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. + /// + public async ValueTask> PeekBatchAsync(int max, CancellationToken ct) { ArgumentOutOfRangeException.ThrowIfNegativeOrZero(max); var batch = new List(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? 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? 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>(batch); + if (droppedHeadTo is long head) + { + _log.TruncateUntil(head); + await _log.CommitAsync(ct).ConfigureAwait(false); // make the head advance durable + } + + return batch; } /// @@ -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 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); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/HistorizationOutboxEntrySerializer.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/HistorizationOutboxEntrySerializer.cs index 60fce25f..9c34b0ca 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/HistorizationOutboxEntrySerializer.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/HistorizationOutboxEntrySerializer.cs @@ -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. /// /// -/// Layout (little-endian): Guid(16) | tagByteLen:int32(4) | tagUtf8(n) | value:double(8) | -/// quality:uint16(2) | timestamp:int64(8). The timestamp is , -/// which round-trips . +/// +/// v2 (written): "HV2\0"(4) | ver=2(1) | Guid(16) | tagByteLen:int32(4) | tagUtf8(n) | +/// kind:byte(1) | payload | status:uint32(4) | timestamp:int64(8). Payload by +/// : Empty none; Double float64; Int64/Boolean/DateTime int64 +/// (Boolean 0/1, DateTime UTC ticks); String int32 byte length + UTF-8. +/// +/// +/// Legacy (read only): Guid(16) | tagByteLen:int32(4) | tagUtf8(n) | value:double(8) | +/// quality:uint16(2) | timestamp:int64(8) — exactly 38 + tagByteLen bytes. Entries +/// sitting in an outbox at upgrade time are read as a Double whose DA quality (always 192) maps +/// to a StatusCode through . +/// +/// +/// Timestamps are , which round-trips . +/// A record matching neither layout is reported unreadable by — +/// never thrown — so a corrupt record cannot jam the drain. +/// /// internal static class HistorizationOutboxEntrySerializer { - private const ushort LegacyGoodQuality = 192; + /// The v2 format version byte that follows the magic. + internal const byte FormatVersion = 2; - /// Serializes to a fixed-layout little-endian byte array. + 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 Magic => "HV2\0"u8; + + /// Serializes in the v2 layout. /// The outbox entry to serialize. - /// The fixed-layout little-endian byte array. + /// The v2 little-endian byte array. 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 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; } - /// Reconstructs a from its serialized bytes. - /// The fixed-layout little-endian bytes produced by . - /// The reconstructed outbox entry. - public static HistorizationOutboxEntry Deserialize(ReadOnlySpan span) + /// + /// Decodes a persisted record: v2 when the magic matches and the v2 structure validates, else + /// the legacy layout when the length is exactly 38 + tagByteLen, else unreadable. + /// Never throws. + /// + /// The persisted record bytes. + /// The decoded entry, or null when unreadable. + /// true when the record decoded under either layout. + public static bool TryDeserialize(ReadOnlySpan 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); + } + + /// Decodes a persisted record, throwing when it matches neither layout. + /// The persisted record bytes. + /// The decoded entry. + /// The record matches neither the v2 nor the legacy layout. + public static HistorizationOutboxEntry Deserialize(ReadOnlySpan 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 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 idBytes) + || !r.TryReadLengthPrefixed(out ReadOnlySpan tagBytes) + || !r.TryTake(1, out ReadOnlySpan 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 d)) + return false; + value = HistorianValue.FromDouble(BinaryPrimitives.ReadDoubleLittleEndian(d)); + break; + case HistorianValueKind.Int64: + if (!r.TryTake(8, out ReadOnlySpan l)) + return false; + value = HistorianValue.FromInt64(BinaryPrimitives.ReadInt64LittleEndian(l)); + break; + case HistorianValueKind.Boolean: + if (!r.TryTake(8, out ReadOnlySpan 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 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 s)) + return false; + value = HistorianValue.FromString(Encoding.UTF8.GetString(s)); + break; + default: + return false; + } + + if (!r.TryTake(4, out ReadOnlySpan status) + || !r.TryTake(8, out ReadOnlySpan 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 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 span, int position) + { + private readonly ReadOnlySpan _span = span; + private int _pos = position; + + public readonly bool AtEnd => _pos == _span.Length; + + public bool TryTake(int count, out ReadOnlySpan 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 slice) + { + slice = default; + return TryTake(4, out ReadOnlySpan len) + && TryTake(BinaryPrimitives.ReadInt32LittleEndian(len), out slice); + } } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/HistorizationOutboxEntrySerializerV2Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/HistorizationOutboxEntrySerializerV2Tests.cs new file mode 100644 index 00000000..d9132e3f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/HistorizationOutboxEntrySerializerV2Tests.cs @@ -0,0 +1,301 @@ +using System.Buffers.Binary; +using System.Text; +using FASTER.core; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.Recorder; + +/// +/// Outbox format v2: every + a real status code round-trips; +/// pre-upgrade (legacy) entries still read; anything else is reported unreadable (never thrown), +/// and the FasterLog outbox skips + counts such a record without letting it pin the head. +/// +public sealed class HistorizationOutboxEntrySerializerV2Tests : IDisposable +{ + // Produced ONCE by the pre-v2 (Task-1 era) Serialize for + // Id=00000000-0000-0000-0000-000000000001, Tag="t", Value=42.5, Quality=192, ts=2026-01-01T00:00:00Z. + // This is what is sitting in a production outbox at upgrade time — never regenerate it. + private const string PinnedLegacyHex = + "0000000000000000000000000000000101000000740000000000404540C0000000F8B4C848DE48"; + + private static readonly DateTime Ts = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + private readonly List _dirs = new(); + + public static TheoryData Kinds() => new() + { + "empty", "double", "nan", "neginf", "int64min", "int64max", "true", "false", + "string", "emptystring", "bigstring", "datetime", + }; + + private static HistorianValue ValueFor(string kind) => kind switch + { + "empty" => HistorianValue.Empty, + "double" => HistorianValue.FromDouble(123.456), + "nan" => HistorianValue.FromDouble(double.NaN), + "neginf" => HistorianValue.FromDouble(double.NegativeInfinity), + "int64min" => HistorianValue.FromInt64(long.MinValue), + "int64max" => HistorianValue.FromInt64(long.MaxValue), + "true" => HistorianValue.FromBoolean(true), + "false" => HistorianValue.FromBoolean(false), + "string" => HistorianValue.FromString("héllo wörld 日本"), + "emptystring" => HistorianValue.FromString(string.Empty), + "bigstring" => HistorianValue.FromString(new string('x', 70 * 1024)), + "datetime" => HistorianValue.FromDateTime(new DateTime(2025, 6, 7, 8, 9, 10, DateTimeKind.Utc)), + _ => throw new ArgumentOutOfRangeException(nameof(kind)), + }; + + [Theory] + [MemberData(nameof(Kinds))] + public void Every_kind_round_trips_with_a_bad_status(string kind) + { + var entry = new HistorizationOutboxEntry(Guid.NewGuid(), "Pumpe_Ω.Temp", ValueFor(kind), 0x80310000u, Ts); + + byte[] bytes = HistorizationOutboxEntrySerializer.Serialize(entry); + HistorizationOutboxEntrySerializer.TryDeserialize(bytes, out HistorizationOutboxEntry? round).ShouldBeTrue(); + + round.ShouldNotBeNull(); + round.Id.ShouldBe(entry.Id); + round.Tag.ShouldBe(entry.Tag); + round.Value.ShouldBe(entry.Value); + round.StatusCode.ShouldBe(0x80310000u); + round.TimestampUtc.ShouldBe(Ts); + round.TimestampUtc.Kind.ShouldBe(DateTimeKind.Utc); + } + + [Fact] + public void V2_layout_is_exactly_as_specified() + { + var id = Guid.Parse("00000000-0000-0000-0000-000000000001"); + byte[] bytes = HistorizationOutboxEntrySerializer.Serialize( + new HistorizationOutboxEntry(id, "t", HistorianValue.FromString("ab"), 0x80310000u, Ts)); + + var expected = new List(); + expected.AddRange("HV2\0"u8.ToArray()); + expected.Add(2); + expected.AddRange(id.ToByteArray()); + expected.AddRange(Le32(1)); + expected.AddRange("t"u8.ToArray()); + expected.Add((byte)HistorianValueKind.String); + expected.AddRange(Le32(2)); + expected.AddRange("ab"u8.ToArray()); + expected.AddRange(Le32(unchecked((int)0x80310000u))); + var ts = new byte[8]; + BinaryPrimitives.WriteInt64LittleEndian(ts, Ts.ToBinary()); + expected.AddRange(ts); + + Convert.ToHexString(bytes).ShouldBe(Convert.ToHexString(expected.ToArray())); + } + + [Fact] + public void Pinned_legacy_bytes_read_as_good_double() + { + byte[] legacy = Convert.FromHexString(PinnedLegacyHex); + + HistorizationOutboxEntrySerializer.TryDeserialize(legacy, out HistorizationOutboxEntry? entry).ShouldBeTrue(); + + entry.ShouldNotBeNull(); + entry.Id.ShouldBe(Guid.Parse("00000000-0000-0000-0000-000000000001")); + entry.Tag.ShouldBe("t"); + entry.Value.Kind.ShouldBe(HistorianValueKind.Double); + entry.Value.Number.ShouldBe(42.5); + entry.StatusCode.ShouldBe(0u); + entry.TimestampUtc.ShouldBe(Ts); + } + + [Fact] + public void A_legacy_record_whose_guid_starts_with_the_v2_magic_still_reads() + { + // Ambiguity guard (#18): a legacy id can begin with the bytes "HV2\0". The v2 structural + // parse fails on it, and the legacy exact-length test must then run before giving up. + byte[] legacy = Convert.FromHexString(PinnedLegacyHex); + "HV2\0"u8.CopyTo(legacy); + legacy[4] = 2; + + HistorizationOutboxEntrySerializer.TryDeserialize(legacy, out HistorizationOutboxEntry? entry).ShouldBeTrue(); + entry!.Value.Number.ShouldBe(42.5); + entry.Tag.ShouldBe("t"); + } + + public static TheoryData Garbage() => new() { "random3", "ver9", "truncated", "legacylength" }; + + [Theory] + [MemberData(nameof(Garbage))] + public void Unreadable_bytes_return_false_and_never_throw(string shape) + { + byte[] bytes = shape switch + { + "random3" => [0x13, 0x37, 0xFF], + "ver9" => WithVersion(9), + "truncated" => Serialize("tag", HistorianValue.FromString("payload"))[..^5], + "legacylength" => [.. Convert.FromHexString(PinnedLegacyHex), 0x00], // 40 != 38 + 1 + _ => throw new ArgumentOutOfRangeException(nameof(shape)), + }; + + HistorizationOutboxEntrySerializer.TryDeserialize(bytes, out HistorizationOutboxEntry? entry).ShouldBeFalse(); + entry.ShouldBeNull(); + } + + [Fact] + public async Task Corrupt_record_between_two_good_ones_is_skipped_counted_and_does_not_block() + { + string dir = NewTempDir(); + var a = Entry("A"); + var c = Entry("C"); + { + using var o = new FasterLogHistorizationOutbox(dir); + await o.AppendAsync(a, TestContext.Current.CancellationToken); + } + + AppendRaw(dir, [0xDE, 0xAD, 0xBE]); + + { + using var o = new FasterLogHistorizationOutbox(dir); + await o.AppendAsync(c, TestContext.Current.CancellationToken); + } + + using var outbox = new FasterLogHistorizationOutbox(dir); + outbox.UnreadableCount.ShouldBe(1); + + var batch = await outbox.PeekBatchAsync(10, TestContext.Current.CancellationToken); + batch.Select(e => e.Id).ShouldBe([a.Id, c.Id]); + outbox.UnreadableCount.ShouldBe(1); // re-peeking a counted record does not count it again + + await outbox.RemoveAsync(c.Id, TestContext.Current.CancellationToken); // FIFO ack passes the garbage + (await outbox.CountAsync(TestContext.Current.CancellationToken)).ShouldBe(0); + (await outbox.PeekBatchAsync(10, TestContext.Current.CancellationToken)).ShouldBeEmpty(); + } + + [Fact] + public async Task Garbage_at_the_head_survives_a_restart_and_does_not_pin_the_head() + { + // Review amendment #1: RecoverState (ctor) must not throw on the garbage, the good entry behind + // it must drain, and the head must advance past the garbage durably. + string dir = NewTempDir(); + AppendRaw(dir, [0x01, 0x02, 0x03, 0x04, 0x05]); + var good = Entry("G"); + { + using var o = new FasterLogHistorizationOutbox(dir); + o.UnreadableCount.ShouldBe(1); + await o.AppendAsync(good, TestContext.Current.CancellationToken); + } + + using (var o = new FasterLogHistorizationOutbox(dir)) // restart: ctor scans the garbage + { + o.UnreadableCount.ShouldBe(1); + var batch = await o.PeekBatchAsync(10, TestContext.Current.CancellationToken); + batch.Select(e => e.Id).ShouldBe([good.Id]); + await o.RemoveAsync(good.Id, TestContext.Current.CancellationToken); + (await o.CountAsync(TestContext.Current.CancellationToken)).ShouldBe(0); + } + + using var reopened = new FasterLogHistorizationOutbox(dir); + reopened.UnreadableCount.ShouldBe(0); // the garbage was truncated away, not re-found + (await reopened.CountAsync(TestContext.Current.CancellationToken)).ShouldBe(0); + } + + [Fact] + public async Task Trailing_garbage_with_no_good_entry_behind_it_is_removed_by_peek() + { + // The pin case: the last good entry is acked (truncating TO the garbage), and nothing readable + // follows, so no RemoveAsync will ever pass it. Peek must drop leading unreadable records. + string dir = NewTempDir(); + var a = Entry("A"); + { + using var o = new FasterLogHistorizationOutbox(dir); + await o.AppendAsync(a, TestContext.Current.CancellationToken); + } + + AppendRaw(dir, [0xAA, 0xBB, 0xCC]); + + using (var o = new FasterLogHistorizationOutbox(dir)) + { + (await o.CountAsync(TestContext.Current.CancellationToken)).ShouldBe(2); + (await o.PeekBatchAsync(10, TestContext.Current.CancellationToken)).Select(e => e.Id).ShouldBe([a.Id]); + await o.RemoveAsync(a.Id, TestContext.Current.CancellationToken); + (await o.CountAsync(TestContext.Current.CancellationToken)).ShouldBe(1); // only the garbage + + (await o.PeekBatchAsync(10, TestContext.Current.CancellationToken)).ShouldBeEmpty(); + (await o.CountAsync(TestContext.Current.CancellationToken)).ShouldBe(0); // head advanced + o.UnreadableCount.ShouldBe(1); + } + + using var reopened = new FasterLogHistorizationOutbox(dir); + (await reopened.CountAsync(TestContext.Current.CancellationToken)).ShouldBe(0); + reopened.UnreadableCount.ShouldBe(0); + } + + [Fact] + public async Task A_pinned_legacy_record_in_the_log_drains_after_upgrade() + { + string dir = NewTempDir(); + AppendRaw(dir, Convert.FromHexString(PinnedLegacyHex)); + + using var o = new FasterLogHistorizationOutbox(dir); + o.UnreadableCount.ShouldBe(0); + var batch = await o.PeekBatchAsync(10, TestContext.Current.CancellationToken); + batch.Count.ShouldBe(1); + batch[0].Value.ShouldBe(HistorianValue.FromDouble(42.5)); + await o.RemoveAsync(batch[0].Id, TestContext.Current.CancellationToken); + (await o.CountAsync(TestContext.Current.CancellationToken)).ShouldBe(0); + } + + private static HistorizationOutboxEntry Entry(string tag) => + new(Guid.NewGuid(), tag, HistorianValue.FromDouble(1), 0u, Ts); + + private static byte[] Serialize(string tag, HistorianValue v) => + HistorizationOutboxEntrySerializer.Serialize(new HistorizationOutboxEntry(Guid.NewGuid(), tag, v, 0u, Ts)); + + private static byte[] WithVersion(byte ver) + { + byte[] b = Serialize("tag", HistorianValue.FromDouble(1)); + b[4] = ver; + return b; + } + + private static byte[] Le32(int v) + { + var b = new byte[4]; + BinaryPrimitives.WriteInt32LittleEndian(b, v); + return b; + } + + // Opens the outbox's FasterLog directly (no outbox) and commits a raw record, as a corrupted or + // pre-upgrade record would sit on disk. + private static void AppendRaw(string dir, byte[] record) + { + using var device = new ManagedLocalStorageDevice(Path.Combine(dir, "hlog.log")); + using var log = new FasterLog(new FasterLogSettings { LogDevice = device }); + log.Enqueue(record); + log.Commit(spinWait: true); + } + + private string NewTempDir() + { + string dir = Path.Combine(Path.GetTempPath(), "histgw-outbox-v2-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(dir); + _dirs.Add(dir); + return dir; + } + + public void Dispose() + { + foreach (string dir in _dirs) + { + try + { + Directory.Delete(dir, recursive: true); + } + catch (IOException) + { + // Best-effort cleanup. + } + catch (UnauthorizedAccessException) + { + // Best-effort cleanup. + } + } + } +}