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

@@ -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,

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;
 /// </summary>
 /// <remarks>
 ///     <para>
-///         <b>Non-throwing by contract.</b> 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 <c>false</c> 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.
+///         <b>Non-throwing by contract, except for cancellation.</b> Any gateway/transport error (and a
+///         non-success, non-queued ack) is mapped to <c>false</c> 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.
 ///     </para>
 ///     <para>
 ///         A success ack OR a store-forward-queued ack maps to <c>true</c>: 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.
 ///     </para>
+///     <para>
+///         <b>Batch normalisation is the SQL backends'</b> (<see cref="LiveWritePreparer.Prepare"/>): 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.
+///     </para>
 /// </remarks>
 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<long>();
             var liveValues = new List<HistorianLiveValue>(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 <remarks>.
-                    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
         }
     }
 
+    /// <summary>The proto live value of a numeric sample (no timestamp), or null for a kind with no double form.</summary>
+    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),
+        };
+    }
+
     /// <summary>
     ///     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.

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 <c>DEDUP UPSERT KEYS(ts, tag)</c> 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).
 /// </summary>
 /// <remarks>
 ///     Batch normalisation is <see cref="LiveWritePreparer.Prepare"/> 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);

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs

@@ -9,8 +9,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
 /// <summary>
 ///     <see cref="IHistorianValueWriter"/> over the SQL <c>Sample</c> table. Redelivery-idempotent: rows are
 ///     inserted only where <c>(TagId, SourceTimeUtc)</c> 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.
 /// </summary>
 /// <remarks>
 ///     <b>Non-finite doubles.</b> NaN / ±Infinity cannot live in a SQL <c>float</c> (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);

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;
+
+/// <summary>
+///     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 (<c>LiveWritePreparer</c>): a Local timestamp is
+///     converted to UTC (it used to be relabelled), non-finite doubles are not sent, and duplicate timestamps keep the first.
+/// </summary>
+public sealed class GatewayValueWriterContractTests
+{
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+    private static GatewayHistorianValueWriter Writer(FakeHistorianGatewayClient fake) => new(fake, NullLogger<GatewayHistorianValueWriter>.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<OperationCanceledException>(() => 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);
+    }
+}

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<OperationCanceledException>(() => 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));

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;
+
+/// <summary>
+///     The recorder's contract: a cancelled drain pass rethrows <see cref="OperationCanceledException"/> out of the writer —
+///     swallowed as <c>false</c> it would be booked as a tag failure and could tip a healthy tag over its attempt budget.
+///     Any other failure is still <c>false</c>. The connection points at a closed port, so nothing reaches a server.
+/// </summary>
+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<OperationCanceledException>(() => 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();
+    }
+}