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
85 lines
3.9 KiB
C#
85 lines
3.9 KiB
C#
using Microsoft.Extensions.Logging;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
|
|
|
|
/// <summary>
|
|
/// The pre-1970 drop Warning is rate-limited per tag (final review M-3) while the counter still counts every dropped
|
|
/// sample. All-pre-1970 batches return before any QuestDB call, so the connection points at a closed port.
|
|
/// </summary>
|
|
public sealed class QuestDbHistorianValueWriterTests
|
|
{
|
|
private static readonly DateTime FileTimeZero = new(1601, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
|
|
|
private static (QuestDbHistorianValueWriter W, ListLogger Log, Action<TimeSpan> Advance) Make()
|
|
{
|
|
var opts = new QuestDbHistorianOptions { ConnectionString = "Host=127.0.0.1;Port=1;Timeout=1" };
|
|
var conn = new QuestDbConnection(opts);
|
|
var boot = new QuestDbSchemaBootstrapper(conn, opts);
|
|
var wal = new QuestDbWalVisibility(conn, opts.VisibilityTimeout);
|
|
var now = new DateTime(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc);
|
|
var log = new ListLogger();
|
|
var w = new QuestDbHistorianValueWriter(conn, new QuestDbTagCatalog(conn, boot, wal), wal, log, () => now);
|
|
return (w, log, d => now += d);
|
|
}
|
|
|
|
private static HistorizationValue[] Pre1970(int n) =>
|
|
Enumerable.Range(0, n).Select(i => new HistorizationValue(FileTimeZero.AddSeconds(i), HistorianValue.FromDouble(i), 0)).ToArray();
|
|
|
|
[Fact]
|
|
public async Task The_pre_1970_warning_is_logged_once_per_tag_per_interval_but_every_drop_is_counted()
|
|
{
|
|
var (w, log, advance) = Make();
|
|
|
|
(await w.WriteLiveValuesAsync("a", Pre1970(2), CancellationToken.None)).ShouldBeTrue();
|
|
advance(TimeSpan.FromSeconds(2));
|
|
(await w.WriteLiveValuesAsync("a", Pre1970(3), CancellationToken.None)).ShouldBeTrue();
|
|
advance(QuestDbHistorianValueWriter.Pre1970WarningInterval - TimeSpan.FromSeconds(3));
|
|
(await w.WriteLiveValuesAsync("a", Pre1970(1), CancellationToken.None)).ShouldBeTrue();
|
|
|
|
log.Warnings.ShouldBe(1);
|
|
w.DroppedPre1970.ShouldBe(6);
|
|
|
|
// A different tag has its own budget.
|
|
(await w.WriteLiveValuesAsync("b", Pre1970(1), CancellationToken.None)).ShouldBeTrue();
|
|
log.Warnings.ShouldBe(2);
|
|
|
|
// Once the interval has passed since the last Warning for "a", it warns again.
|
|
advance(TimeSpan.FromSeconds(1));
|
|
(await w.WriteLiveValuesAsync("a", Pre1970(1), CancellationToken.None)).ShouldBeTrue();
|
|
log.Warnings.ShouldBe(3);
|
|
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));
|
|
|
|
private sealed class ListLogger : ILogger
|
|
{
|
|
private int _warnings;
|
|
public int Warnings => Volatile.Read(ref _warnings);
|
|
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
|
public bool IsEnabled(LogLevel logLevel) => true;
|
|
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
|
|
{
|
|
if (logLevel == LogLevel.Warning) Interlocked.Increment(ref _warnings);
|
|
}
|
|
}
|
|
}
|