From 7c5cc9c7e5bf7723978dc19279f77a9a7f65f20b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 16:42:13 -0400 Subject: [PATCH] fix(historian): live writers rethrow pass cancellation; gateway writes normalised (lmxopcua#746) Every IHistorianValueWriter caught Exception and returned false, including OperationCanceledException from the drain's own token. The recorder's guard ("a cancelled pass is not a tag failure") was therefore dead code: at shutdown mid-pass each cancelled group could be booked as a consecutive failure and, with a sibling that already wrote, tip a healthy tag toward quarantine. SqlServer, QuestDb and Gateway writers now rethrow a cancellation of the caller's token; a cancellation-shaped error on a live token is still an ordinary failed write. The Gateway writer relabelled a Local-kind timestamp as UTC instead of converting it, so the same value could land at two different times on two backends. Stamped samples now go through LiveWritePreparer (Local -> UTC, non-finite doubles not sent, duplicate timestamps keep the first); unstamped samples are still left for the gateway to stamp, in batch order. Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22 in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV --- docs/Historian.md | 7 ++ .../Recorder/GatewayHistorianValueWriter.cs | 77 +++++++++++------ .../QuestDbHistorianValueWriter.cs | 7 +- .../SqlServerHistorianValueWriter.cs | 9 +- .../GatewayValueWriterContractTests.cs | 83 +++++++++++++++++++ .../QuestDbHistorianValueWriterTests.cs | 13 +++ .../SqlServerValueWriterCancellationTests.cs | 37 +++++++++ 7 files changed, 204 insertions(+), 29 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/GatewayValueWriterContractTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerValueWriterCancellationTests.cs diff --git a/docs/Historian.md b/docs/Historian.md index 6bf5dd59e..3da225773 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -219,6 +219,13 @@ source it from there. > it is sent (a server-managed column; 0.2.0 Contracts C-002), so **AVEVA still stores every recorded value as > Good** — a pre-existing limitation of that backend, now documented rather than fixed (archreview 06/C-7). The > `SqlServer` backend stores the real status. +> +> **Live writes are normalised alike on every backend.** The Gateway writer now runs each stamped sample through the SQL +> backends' `LiveWritePreparer`: a `Local`-kind timestamp is **converted** to UTC (it used to be relabelled UTC, shifting it by +> the node's offset), NaN / ±Infinity are not sent (the SQL backends store them as an Empty + Bad sample; the gateway's live +> path is numeric-only), and a duplicate timestamp within a batch keeps the first. An unstamped sample is still left for the +> gateway to stamp. Every live writer returns `false` for a failed write but **rethrows a cancellation of the drain pass** +> (shutdown), so a node stop can never be booked as a tag failure toward the #538 quarantine. > ✅ **Continuous value capture is LIVE-PROVEN END-TO-END > ([#491](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/491) CLOSED — gate run 2026-08-01, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/GatewayHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/GatewayHistorianValueWriter.cs index de7f3bd78..8a18d2484 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/GatewayHistorianValueWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/GatewayHistorianValueWriter.cs @@ -2,6 +2,7 @@ using Google.Protobuf.WellKnownTypes; using Microsoft.Extensions.Logging; using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; @@ -14,11 +15,11 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; /// /// /// -/// Non-throwing by contract. The recorder's drain loop stays simple by treating the -/// writer as never throwing: any gateway/transport error (and a non-success, non-queued ack) -/// is mapped to false so the recorder retains the outbox entries and retries. Only the -/// failure category (the exception type name) is logged — never tag values, hostnames, or -/// credentials. +/// Non-throwing by contract, except for cancellation. Any gateway/transport error (and a +/// non-success, non-queued ack) is mapped to false so the recorder retains the outbox entries and +/// retries; a cancellation of the caller's token is rethrown (it ends the drain pass and must not count as +/// this tag's failure). Only the failure category (the exception type name) is logged — never tag values, +/// hostnames, or credentials. /// /// /// A success ack OR a store-forward-queued ack maps to true: a value the gateway @@ -41,6 +42,11 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; /// never thrown; a batch with none left is acked without calling the gateway. The recorder already drops /// those kinds at capture for this backend, so the skip is a backstop. /// +/// +/// Batch normalisation is the SQL backends' (): a Local +/// timestamp is converted to UTC, NaN / ±Infinity are not sent (they become Empty), and duplicate timestamps in +/// a batch keep the first. An unstamped value is left unstamped for the gateway to stamp. +/// /// public sealed class GatewayHistorianValueWriter : IHistorianValueWriter, IAsyncDisposable { @@ -71,47 +77,47 @@ public sealed class GatewayHistorianValueWriter : IHistorianValueWriter, IAsyncD try { + // The SQL backends' batch normalisation (LiveWritePreparer): a Local timestamp is converted to UTC (it used to be + // relabelled, shifting it by the node's offset), a non-finite double becomes Empty (so it is not sent — the live + // path is numeric-only), and a duplicate timestamp keeps the first. An UNSTAMPED value stays unstamped — the + // gateway stamps it — so each stamped value is normalised on its own (Prepare would stamp a null with the node's + // clock) and the batch keeps its order. + var seen = new HashSet(); var liveValues = new List(values.Count); foreach (HistorizationValue value in values) { - if (!value.Value.TryGetDouble(out double number)) + if (value.TimestampUtc is null) { - // The gateway live path is numeric-only; non-numeric kinds are skipped, never thrown. - _logger.LogDebug("Skipping non-numeric {Kind} value for live write.", value.Value.Kind); + if (ToLive(value.Value, value.StatusCode) is { } unstamped) liveValues.Add(unstamped); continue; } - var live = new HistorianLiveValue - { - NumericValue = number, - // Documented-dead input on the WriteLiveValues SQL path (0.2.0 Contracts C-002 — - // server-managed, silently discarded). Populated deliberately from the value's - // status; Good maps to 192 as before. See the class . - Quality = Mapping.GatewayQualityMapper.ToDaQuality(value.StatusCode), - }; - - if (value.TimestampUtc is { } timestampUtc) - { - // Timestamp.FromDateTime requires Utc kind; coerce defensively. A null timestamp - // leaves the proto field unset -> the gateway's SQL writer server-stamps now. - live.Timestamp = Timestamp.FromDateTime(DateTime.SpecifyKind(timestampUtc, DateTimeKind.Utc)); - } - + var row = LiveWritePreparer.Prepare([value], static () => DateTime.UtcNow)[0]; + if (!seen.Add(row.TimeUtc.Ticks)) continue; // first wins, as on the SQL backends + if (ToLive(row.Value, row.StatusCode) is not { } live) continue; + live.Timestamp = Timestamp.FromDateTime(row.TimeUtc); // Utc kind (Prepare normalises it) liveValues.Add(live); } if (liveValues.Count == 0) { - // Every value was non-numeric: nothing to send, trivially drained. + // Every value was non-numeric (or non-finite): nothing to send, trivially drained. return true; } WriteAck ack = await _client.WriteLiveValuesAsync(tag, liveValues, ct).ConfigureAwait(false); return ack.Success || ack.Queued; } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + // The drain pass was cancelled (shutdown): rethrow — reported as false it would be booked as THIS tag's + // failure and could tip a healthy tag over its attempt budget. A cancellation-shaped error on a live token + // (the gateway's own deadline) falls through to the catch-all: an ordinary failed write. + throw; + } catch (Exception exception) { - // NEVER throw out of the writer — the recorder's drain expects a bool so its retain/retry + // Otherwise never throw out of the writer — the recorder's drain expects a bool so its retain/retry // logic stays simple. Log only the failure category (no value content, hostnames, or creds). _logger.LogDebug( "WriteLiveValues failed ({Exception}); recorder will retain and retry.", @@ -120,6 +126,25 @@ public sealed class GatewayHistorianValueWriter : IHistorianValueWriter, IAsyncD } } + /// The proto live value of a numeric sample (no timestamp), or null for a kind with no double form. + private HistorianLiveValue? ToLive(HistorianValue value, uint statusCode) + { + if (!value.TryGetDouble(out double number) || !double.IsFinite(number)) + { + // The gateway live path is numeric-only; non-numeric kinds (and non-finite doubles) are skipped, never thrown. + _logger.LogDebug("Skipping non-numeric {Kind} value for live write.", value.Kind); + return null; + } + + return new HistorianLiveValue + { + NumericValue = number, + // Documented-dead input on the WriteLiveValues SQL path (0.2.0 Contracts C-002 — server-managed, silently + // discarded). Populated deliberately from the value's status; Good maps to 192. See the class remarks. + Quality = Mapping.GatewayQualityMapper.ToDaQuality(statusCode), + }; + } + /// /// Disposes the underlying gateway client (and its gRPC channel). The DI container owns this /// writer as a singleton, so this fires once at host shutdown — closing the channel gracefully. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs index c9eb37d02..3f537daf2 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs @@ -14,7 +14,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// semantics: QuestDB's DEDUP UPSERT KEYS(ts, tag) would let a re-sent sample overwrite the stored one (and a /// re-sent live sample resurrect a tombstone), so the writer reads which timestamps already exist for the tag in /// the batch's range (tombstones count as present, T0 Q11 range form) and inserts only the absent rows. -/// Never throws; returns false on any failure so the recorder keeps its entries. +/// Returns false on any failure so the recorder keeps its entries — except a cancellation of the caller's token, which +/// is rethrown (it ends the pass; it is not this tag's failure). /// /// /// Batch normalisation is at µs resolution (non-finite doubles become Empty @@ -101,6 +102,10 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); // timeout/suspension does not change the result (R5) return true; } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; // the drain pass was cancelled (shutdown) — not a write failure of this tag + } catch (Exception ex) { _log.LogWarning("QuestDB historian live-value write failed ({Exception}); will retry.", ex.GetType().Name); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs index 8ad105129..1dc58c395 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs @@ -9,8 +9,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; /// /// over the SQL Sample table. Redelivery-idempotent: rows are /// inserted only where (TagId, SourceTimeUtc) is absent, so the recorder's at-least-once redelivery -/// leaves the first-stored sample untouched. Never throws; returns false on any failure so the recorder keeps -/// the entries and its per-tag quarantine keeps working. +/// leaves the first-stored sample untouched. Returns false on any failure so the recorder keeps the entries and its +/// per-tag quarantine keeps working — except a cancellation of the caller's token, which is rethrown: it ends the +/// PASS, and booked as a tag failure it could tip a healthy tag over its attempt budget. /// /// /// Non-finite doubles. NaN / ±Infinity cannot live in a SQL float (the TVP throws). They are stored @@ -90,6 +91,10 @@ internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter } } } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; // the drain pass was cancelled (shutdown) — not a write failure of this tag + } catch (Exception ex) { _log.LogWarning("SQL historian live-value write failed ({Exception}); will retry.", ex.GetType().Name); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/GatewayValueWriterContractTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/GatewayValueWriterContractTests.cs new file mode 100644 index 000000000..6580b670c --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/GatewayValueWriterContractTests.cs @@ -0,0 +1,83 @@ +using Microsoft.Extensions.Logging.Abstractions; +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; + +/// +/// The gateway live writer keeps the recorder's cancellation contract (a cancellation of the PASS rethrows; it is not a +/// tag failure) and normalises a batch the way the SQL backends do (LiveWritePreparer): a Local timestamp is +/// converted to UTC (it used to be relabelled), non-finite doubles are not sent, and duplicate timestamps keep the first. +/// +public sealed class GatewayValueWriterContractTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + private static GatewayHistorianValueWriter Writer(FakeHistorianGatewayClient fake) => new(fake, NullLogger.Instance); + private static readonly HistorizationValue[] One = [new(DateTime.UtcNow, HistorianValue.FromDouble(1), 0)]; + + [Fact] + public async Task A_cancellation_of_the_callers_token_is_rethrown_not_reported_as_a_failed_write() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + var fake = new FakeHistorianGatewayClient { WriteLiveValuesThrows = new OperationCanceledException(cts.Token) }; + + await Should.ThrowAsync(() => Writer(fake).WriteLiveValuesAsync("T", One, cts.Token)); + } + + [Fact] + public async Task A_cancellation_shaped_failure_on_a_live_token_is_an_ordinary_failed_write() + { + // The gateway's own internal deadline, say: the pass was not cancelled, so the tag's write failed (false), and the + // recorder's quarantine can count it. + var fake = new FakeHistorianGatewayClient { WriteLiveValuesThrows = new OperationCanceledException() }; + (await Writer(fake).WriteLiveValuesAsync("T", One, Ct)).ShouldBeFalse(); + } + + [Fact] + public async Task A_local_timestamp_is_converted_to_utc_not_relabelled() + { + var fake = new FakeHistorianGatewayClient(); + var local = new DateTime(2026, 10, 22, 9, 30, 0, DateTimeKind.Local); + + await Writer(fake).WriteLiveValuesAsync("T", [new HistorizationValue(local, HistorianValue.FromDouble(1), 0)], Ct); + + fake.LastWriteLiveValues!.Single().Timestamp!.ToDateTime().ShouldBe(local.ToUniversalTime()); + } + + [Fact] + public async Task Non_finite_doubles_are_not_sent_and_duplicate_timestamps_keep_the_first() + { + var fake = new FakeHistorianGatewayClient(); + var t = new DateTime(2026, 10, 22, 8, 0, 0, DateTimeKind.Utc); + HistorizationValue[] batch = + [ + new(t, HistorianValue.FromDouble(1), 0), + new(t, HistorianValue.FromDouble(2), 0), // duplicate time ⇒ dropped (first wins) + new(t.AddSeconds(1), HistorianValue.FromDouble(double.NaN), 0), // non-finite ⇒ Empty ⇒ not numeric ⇒ not sent + new(t.AddSeconds(2), HistorianValue.FromDouble(double.PositiveInfinity), 0), + new(t.AddSeconds(3), HistorianValue.FromDouble(3), 0), + new(null, HistorianValue.FromDouble(double.NaN), 0), // unstamped non-finite ⇒ not sent either + new(null, HistorianValue.FromDouble(4), 0), // unstamped ⇒ still left for the gateway to stamp + ]; + + (await Writer(fake).WriteLiveValuesAsync("T", batch, Ct)).ShouldBeTrue(); + + var sent = fake.LastWriteLiveValues!; + sent.Select(v => v.NumericValue).ShouldBe([1d, 3d, 4d]); + sent[0].Timestamp!.ToDateTime().ShouldBe(t); + sent[1].Timestamp!.ToDateTime().ShouldBe(t.AddSeconds(3)); + sent[2].Timestamp.ShouldBeNull(); + } + + [Fact] + public async Task A_batch_of_only_non_finite_values_is_acked_without_calling_the_gateway() + { + var fake = new FakeHistorianGatewayClient(); + (await Writer(fake).WriteLiveValuesAsync("T", + [new HistorizationValue(DateTime.UtcNow, HistorianValue.FromDouble(double.NaN), 0)], Ct)).ShouldBeTrue(); + fake.WriteLiveValuesCallCount.ShouldBe(0); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianValueWriterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianValueWriterTests.cs index 6650188e0..a5b846105 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianValueWriterTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianValueWriterTests.cs @@ -53,6 +53,19 @@ public sealed class QuestDbHistorianValueWriterTests w.DroppedPre1970.ShouldBe(8); } + [Fact] + public async Task A_cancelled_pass_rethrows_rather_than_reporting_a_failed_write() + { + var (w, _, _) = Make(); + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + HistorizationValue[] one = [new(new DateTime(2026, 10, 22, 0, 0, 0, DateTimeKind.Utc), HistorianValue.FromDouble(1), 0)]; + await Should.ThrowAsync(() => w.WriteLiveValuesAsync("a", one, cts.Token)); + + // Positive control: the same unreachable server on a live token is an ordinary failed write. + (await w.WriteLiveValuesAsync("a", one, TestContext.Current.CancellationToken)).ShouldBeFalse(); + } + [Fact] public void The_interval_is_ten_minutes() => QuestDbHistorianValueWriter.Pre1970WarningInterval.ShouldBe(TimeSpan.FromMinutes(10)); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerValueWriterCancellationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerValueWriterCancellationTests.cs new file mode 100644 index 000000000..5f06acd2d --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerValueWriterCancellationTests.cs @@ -0,0 +1,37 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests; + +/// +/// The recorder's contract: a cancelled drain pass rethrows out of the writer — +/// swallowed as false it would be booked as a tag failure and could tip a healthy tag over its attempt budget. +/// Any other failure is still false. The connection points at a closed port, so nothing reaches a server. +/// +public sealed class SqlServerValueWriterCancellationTests +{ + private static SqlServerHistorianValueWriter Writer() + { + var opts = new SqlServerHistorianOptions { ConnectionString = "Server=127.0.0.1,1;Connect Timeout=1;TrustServerCertificate=True" }; + var conn = new SqlHistorianConnection(opts); + var boot = new SqlServerSchemaBootstrapper(conn, opts); + return new SqlServerHistorianValueWriter(conn, boot, new SqlServerTagCatalog(conn, boot)); + } + + private static readonly HistorizationValue[] One = [new(DateTime.UtcNow, HistorianValue.FromDouble(1), 0)]; + + [Fact] + public async Task A_cancelled_pass_rethrows() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + await Should.ThrowAsync(() => Writer().WriteLiveValuesAsync("T", One, cts.Token)); + } + + [Fact] + public async Task An_unreachable_server_is_a_failed_write_not_a_throw() + { + (await Writer().WriteLiveValuesAsync("T", One, TestContext.Current.CancellationToken)).ShouldBeFalse(); + } +}