Joseph Doherty
7c5cc9c7e5
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
84 lines
4.1 KiB
C#
84 lines
4.1 KiB
C#
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);
|
|
}
|
|
}
|