Merge master into feat/677-aggregates

# Conflicts:
#	docs/Historian.md
This commit is contained in:
Joseph Doherty
2026-10-11 11:30:46 -04:00
66 changed files with 2054 additions and 154 deletions
@@ -0,0 +1,106 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
/// <summary>
/// The direction-aware event read (Gitea #675): forward is <c>[lo, hi)</c> ascending by <c>(EventTimeUtc, EventId)</c>;
/// reverse is the Part 11 reverse window <c>(lo, hi]</c> DESCENDING by the same key, so a capped reverse read returns the
/// LATEST events and flags truncation. Paging a reverse read the way the server does — resume at the last returned
/// time inclusive and drop the boundary ties already emitted — must visit every event exactly once, newest-first,
/// through a same-timestamp tie cluster.
/// </summary>
public abstract class EventReadConformance : ConformanceBase
{
private static readonly CancellationToken Ct = CancellationToken.None;
/// <summary>e0@0, e1@1, then the ties t0..t4@2, then e5@3 and e6@4 (all source "s"), plus o0@2 from another source.</summary>
private async Task<IHistorianDataSource> SeedAsync()
{
await B.SeedEventsAsync(
(At(0), "e0", "s"), (At(1), "e1", "s"),
(At(2), "t0", "s"), (At(2), "t1", "s"), (At(2), "t2", "s"), (At(2), "t3", "s"), (At(2), "t4", "s"),
(At(3), "e5", "s"), (At(4), "e6", "s"),
(At(2), "o0", "other"));
return B.CreateDataSource();
}
private static readonly string[] NewestFirst = ["e6", "e5", "t4", "t3", "t2", "t1", "t0", "e1", "e0"];
[Fact]
public async Task Reverse_read_is_newest_first_by_time_then_EventId_descending()
{
using var src = await SeedAsync();
var r = await src.ReadEventsAsync("s", At(-1), At(10), 0, reverse: true, Ct);
r.Events.Select(e => e.EventId).ShouldBe(NewestFirst);
r.ContinuationPoint.ShouldBeNull();
}
[Fact]
public async Task Reverse_window_is_end_exclusive_start_inclusive_whichever_order_the_bounds_come_in()
{
using var src = await SeedAsync();
// (At(1), At(3)]: e1 at the lower bound is excluded, e5 at the upper bound included.
(await src.ReadEventsAsync("s", At(3), At(1), 0, reverse: true, Ct)).Events.Select(e => e.EventId)
.ShouldBe(["e5", "t4", "t3", "t2", "t1", "t0"]);
(await src.ReadEventsAsync("s", At(1), At(3), 0, reverse: true, Ct)).Events.Select(e => e.EventId)
.ShouldBe(["e5", "t4", "t3", "t2", "t1", "t0"]);
// Forward is unchanged: [At(1), At(3)) ascending.
(await src.ReadEventsAsync("s", At(1), At(3), 0, reverse: false, Ct)).Events.Select(e => e.EventId)
.ShouldBe(["e1", "t0", "t1", "t2", "t3", "t4"]);
}
[Fact]
public async Task A_capped_reverse_read_keeps_the_LATEST_events_and_flags_truncation()
{
using var src = await SeedAsync();
var r = await src.ReadEventsAsync("s", At(10), At(-1), 3, reverse: true, Ct);
r.Events.Select(e => e.EventId).ShouldBe(["e6", "e5", "t4"]);
r.ContinuationPoint.ShouldNotBeNull();
// An exact fit is complete.
(await src.ReadEventsAsync("s", At(10), At(-1), 9, reverse: true, Ct)).ContinuationPoint.ShouldBeNull();
}
[Fact]
public async Task Paging_a_reverse_read_through_a_tie_cluster_visits_every_event_once_newest_first()
{
using var src = await SeedAsync();
foreach (var cap in new[] { 1, 2, 3, 4 })
{
var all = new List<string>();
var lo = At(-1);
var from = At(10);
var skip = 0;
for (var page = 0; ; page++)
{
page.ShouldBeLessThan(40, "paging must terminate");
var r = await src.ReadEventsAsync("s", from, lo, cap + skip, reverse: true, Ct);
var events = r.Events.ToList();
var drop = 0;
while (drop < skip && drop < events.Count && events[drop].EventTimeUtc == from) drop++;
events = events.Skip(drop).ToList();
// The server asks for cap + skip rows (it re-reads the boundary ties it then drops), so a page is never
// emptied by the trim while more remains.
if (r.ContinuationPoint is not null) events.Count.ShouldBe(cap);
all.AddRange(events.Select(e => e.EventId));
if (r.ContinuationPoint is null) break;
var boundary = events[^1].EventTimeUtc;
var ties = events.AsEnumerable().Reverse().TakeWhile(e => e.EventTimeUtc == boundary).Count();
skip = boundary == from ? skip + ties : ties;
from = boundary;
}
all.SequenceEqual(NewestFirst).ShouldBeTrue($"cap {cap}: {string.Join(",", all)}");
}
}
[Fact]
public async Task Reverse_read_with_a_null_source_spans_every_source()
{
using var src = await SeedAsync();
var r = await src.ReadEventsAsync(null, At(2.5), At(1.5), 0, reverse: true, Ct);
r.Events.Select(e => e.EventId).ShouldBe(["t4", "t3", "t2", "t1", "t0", "o0"]);
}
}
@@ -41,6 +41,13 @@ public interface IConformanceBackend : IAsyncDisposable
/// <param name="samples">The rows to store.</param>
Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples);
/// <summary>
/// Writes historical events directly in the backend's event table (never through the alarm writer): event id,
/// source and time as given (rounded to the storage resolution), severity 500, received time = event time.
/// </summary>
/// <param name="events">The events to store.</param>
Task SeedEventsAsync(params (DateTime T, string Id, string Source)[] events);
/// <summary>
/// Live (non-deleted) rows for a tag, ascending by time, read directly from storage. The status is returned
/// VERBATIM as stored — there is no read-side clamp (an Empty row stored non-Bad reads back non-Bad here).
@@ -71,7 +71,7 @@ public sealed class WireBackendTests
await driver.InitializeAsync("{}", ct);
await WaitFor(() =>
driver.GetDeviceState(DeviceHost) is { FixedTreeCache: not null }, TimeSpan.FromSeconds(5));
driver.GetDeviceState(DeviceHost) is { FixedTreeCache: not null }, WaitTimeout);
var state = driver.GetDeviceState(DeviceHost);
state.ShouldNotBeNull();
@@ -82,7 +82,7 @@ public sealed class WireBackendTests
await WaitFor(() =>
state.LastFixedSnapshots.ContainsKey($"{DeviceHost}/Axes/X/AbsolutePosition"),
TimeSpan.FromSeconds(3));
WaitTimeout);
state.LastFixedSnapshots[$"{DeviceHost}/Axes/X/AbsolutePosition"].ShouldBe(123456);
}
}
@@ -136,7 +136,7 @@ public sealed class WireBackendTests
await WaitFor(() =>
driver.GetDeviceState(DeviceHost) is { LastProgramInfo: not null },
TimeSpan.FromSeconds(5));
WaitTimeout);
var snapshots = await driver.ReadAsync(
[$"{DeviceHost}/Program/Name",
@@ -194,7 +194,7 @@ public sealed class WireBackendTests
{
var state = driver.GetDeviceState(DeviceHost);
return state is not null && state.LastTimers.Count == 4;
}, TimeSpan.FromSeconds(5));
}, WaitTimeout);
var snapshots = await driver.ReadAsync(
[$"{DeviceHost}/Timers/PowerOnSeconds",
@@ -265,8 +265,11 @@ public sealed class WireBackendTests
{
var state = driver.GetDeviceState(DeviceHost);
return state?.FixedTreeCache is { Capabilities.SpindleLoad: true, Capabilities.SpindleMotorSpeed: true }
&& state.LastSpindleLoads.Count >= 2;
}, TimeSpan.FromSeconds(5));
&& state.LastSpindleLoads.Count >= 2
// Motor speeds are cached after a SECOND wire round-trip (FocasDriver spindle poll), so the loads
// arriving first does not mean the speeds have — waiting on loads alone raced under load (#678.3).
&& state.LastSpindleMotorSpeeds.Count >= 2;
}, WaitTimeout);
var snapshots = await driver.ReadAsync(
[$"{DeviceHost}/Spindle/S1/Load",
@@ -282,6 +285,10 @@ public sealed class WireBackendTests
}
}
/// <summary>Generous: the sim fixture answers fast, but a loaded CI host (e.g. a concurrent docker build) does not.</summary>
private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(15);
/// <summary>Polls <paramref name="pred"/>; FAILS the test with a clear message on timeout rather than returning silently.</summary>
private static async Task WaitFor(Func<bool> pred, TimeSpan timeout)
{
var deadline = DateTime.UtcNow + timeout;
@@ -290,5 +297,7 @@ public sealed class WireBackendTests
if (pred()) return;
await Task.Delay(50);
}
if (pred()) return;
Assert.Fail($"Condition not met within {timeout.TotalSeconds:0.#}s waiting on the FOCAS wire backend.");
}
}
@@ -0,0 +1,123 @@
using System.Net;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
/// <summary>
/// Gitea #675: the Gateway backend's event-read bounds and direction, offline.
/// <list type="bullet">
/// <item>A reverse (newest-first) read is refused before anything is sent — the gateway streams ascending only.</item>
/// <item>The open upper bound a start-only forward read uses (<c>DateTime.MaxValue</c>, UTC kind) survives the
/// PUBLISHED client package's proto conversion — captured on the wire by an in-process gRPC server — and the
/// Windows FILETIME encoding the gateway converts it to. The live AVEVA leg (what the historian does with a
/// year-9999 bound) stays open: it needs the VPN.</item>
/// </list>
/// </summary>
public sealed class GatewayEventBoundsTests
{
private static readonly DateTime UtcMax = new(DateTime.MaxValue.Ticks, DateTimeKind.Utc);
private static CancellationToken Ct => TestContext.Current.CancellationToken;
[Fact]
public async Task A_reverse_events_read_is_refused_and_sends_nothing()
{
var fake = new FakeHistorianGatewayClient();
IHistorianDataSource ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
await Should.ThrowAsync<NotSupportedException>(() =>
ds.ReadEventsAsync("Pump1", DateTime.UtcNow, DateTime.UtcNow.AddHours(-1), 10, reverse: true, Ct));
fake.ReadEventsCallCount.ShouldBe(0);
// Forward through the same overload is the five-argument read, unchanged.
await ds.ReadEventsAsync("Pump1", DateTime.UtcNow.AddHours(-1), DateTime.UtcNow, 10, reverse: false, Ct);
fake.ReadEventsCallCount.ShouldBe(1);
}
[Fact]
public void UtcMax_round_trips_through_the_proto_Timestamp_and_FILETIME_encodings()
{
var ts = Timestamp.FromDateTime(UtcMax);
ts.ToDateTime().ShouldBe(UtcMax);
ts.ToDateTime().Kind.ShouldBe(DateTimeKind.Utc);
// The gateway hands its bounds to the historian as FILETIME (100 ns since 1601) — in range up to year 30828.
var fileTime = ts.ToDateTime().ToFileTimeUtc();
DateTime.FromFileTimeUtc(fileTime).ShouldBe(UtcMax);
}
[Fact]
public async Task A_start_only_bound_of_UtcMax_reaches_the_wire_intact_through_the_published_client()
{
var capture = new CapturingReadService();
await using var server = await StartServerAsync(capture);
using var adapter = HistorianGatewayClientAdapter.Create(
new ServerHistorianOptions
{
Enabled = true,
Endpoint = $"http://127.0.0.1:{server.Port}",
ApiKey = "histgw_test_key",
UseTls = false,
},
NullLoggerFactory.Instance);
IHistorianDataSource ds = new GatewayHistorianDataSource(adapter, NullLogger<GatewayHistorianDataSource>.Instance);
var start = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
var r = await ds.ReadEventsAsync("Pump1", start, UtcMax, 5, Ct);
r.Events.ShouldBeEmpty();
capture.Request.ShouldNotBeNull();
capture.Request!.StartTime.ToDateTime().ShouldBe(start);
capture.Request.EndTime.ToDateTime().ShouldBe(UtcMax);
DateTime.FromFileTimeUtc(capture.Request.EndTime.ToDateTime().ToFileTimeUtc()).ShouldBe(UtcMax);
}
private sealed class CapturingReadService : HistorianRead.HistorianReadBase
{
public ReadEventsRequest? Request { get; private set; }
public override Task ReadEvents(
ReadEventsRequest request, IServerStreamWriter<HistorianEvent> responseStream, ServerCallContext context)
{
Request = request;
return Task.CompletedTask;
}
}
private sealed class Server(WebApplication app, int port) : IAsyncDisposable
{
public int Port { get; } = port;
public async ValueTask DisposeAsync()
{
await app.StopAsync();
await app.DisposeAsync();
}
}
private static async Task<Server> StartServerAsync(CapturingReadService service)
{
var builder = WebApplication.CreateSlimBuilder();
builder.Logging.ClearProviders();
builder.WebHost.ConfigureKestrel(k =>
k.Listen(IPAddress.Loopback, 0, o => o.Protocols = HttpProtocols.Http2));
builder.Services.AddGrpc();
builder.Services.AddSingleton(service);
var app = builder.Build();
app.MapGrpcService<CapturingReadService>();
await app.StartAsync();
var port = new Uri(app.Urls.First()).Port;
return new Server(app, port);
}
}
@@ -15,12 +15,18 @@
<PackageReference Include="Microsoft.NET.Test.Sdk"/>
<!-- ServiceCollection + BuildServiceProvider for the CreateDataSource factory test. -->
<PackageReference Include="Microsoft.Extensions.DependencyInjection"/>
<!-- In-process gRPC server capturing the published client's ReadEvents request (#675 bounds pin). -->
<PackageReference Include="Grpc.AspNetCore"/>
<PackageReference Include="xunit.runner.visualstudio">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/>
</ItemGroup>
@@ -60,3 +60,10 @@ public sealed class QuestDbModifiedValueConformance(QuestDbFixture fx) : Modifie
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbEventReadConformance(QuestDbFixture fx) : EventReadConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
@@ -143,6 +143,27 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend
await Wal.WaitAsync(Conn.SampleTable, CancellationToken.None);
}
/// <inheritdoc/>
public async Task SeedEventsAsync(params (DateTime T, string Id, string Source)[] events)
{
await using (var c = await Conn.OpenAsync(CancellationToken.None))
{
foreach (var (t, id, source) in events)
{
await using var cmd = Conn.CreateCommand(c,
$"INSERT INTO {Conn.EventTable} (ts, event_id, source, severity, message, received_utc) VALUES ($1, $2, $3, $4, $5, $6)");
QuestDbConnection.BindTimestamp(cmd, t);
QuestDbConnection.BindText(cmd, id);
QuestDbConnection.BindText(cmd, source);
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = 500 });
QuestDbConnection.BindText(cmd, null);
QuestDbConnection.BindTimestamp(cmd, t);
await cmd.ExecuteNonQueryAsync();
}
}
await Wal.WaitAsync(Conn.EventTable, CancellationToken.None);
}
/// <summary>Runs a statement against this table set (test-only escape hatch for hand-made rows), then waits for WAL visibility of the sample table.</summary>
/// <param name="sql">The statement; parameters are bound positionally by <paramref name="bind"/>.</param>
/// <param name="bind">Binds the parameters.</param>
@@ -0,0 +1,80 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
/// <summary>
/// Gitea #674.1 — pins the QuestDB TTL behaviour the historian's retention relies on: a table TTL is judged against
/// the wall clock (<c>min(max timestamp, now)</c>), so one far-future row — a 2066 typo that reached the backend
/// without going through the node manager's <c>HistoryUpdateMaxFutureSkew</c> bound — cannot age out every older
/// partition, including the Part 11 modification trail (the record of who changed history).
/// </summary>
/// <remarks>
/// <para>
/// <b>Positive control.</b> A partition genuinely older than the TTL (40 days, against a 30-day TTL) is kept until
/// the far-future commit and evicted BY it — proving TTL housekeeping ran on exactly the commit that could have
/// dropped the in-TTL partitions. Without it the "nothing evicted" assertion would also pass against a table
/// whose TTL never ran.
/// </para>
/// <para>
/// ⚠️ Observed on QuestDB 10.0.1 (2026-10-11): the in-TTL partitions also survived with
/// <c>cairo.ttl.use.wall.clock=false</c> (verified effective via <c>SHOW PARAMETERS</c>), so this test does not
/// isolate that flag — it pins the outcome. Keep the flag at its default (<c>true</c>) regardless; the server-side
/// future-skew bound is the primary guard.
/// </para>
/// </remarks>
[Collection(QuestDbCollection.Name)]
public sealed class TtlWallClockTests(QuestDbFixture fx)
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static readonly HistoryUpdateContext Who = new("ttl-test", null);
[Fact]
public async Task A_far_future_row_evicts_only_partitions_older_than_the_ttl_by_the_wall_clock()
{
fx.SkipIfUnavailable();
var o = fx.Options(fx.UniquePrefix(), retentionDays: 30);
var c = new QuestDbConnection(o);
var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout);
var tags = new QuestDbTagCatalog(c, new QuestDbSchemaBootstrapper(c, o), wal);
var updater = new QuestDbHistorianDataUpdater(c, tags, wal);
// Whole seconds (µs-aligned), one row per DAY partition. With the newest row 10 days old, the 40-day partition is
// still inside the TTL measured from the table's max timestamp, so this commit evicts nothing.
var now = DateTime.UtcNow;
var today = new DateTime(now.Year, now.Month, now.Day, now.Hour, now.Minute, now.Second, DateTimeKind.Utc);
DateTime[] older = [today.AddDays(-40), today.AddDays(-20), today.AddDays(-10)];
(await updater.UpdateDataAsync("ttl.tag", HistoryUpdateMode.Insert, older.Select((t, i) => V(t, i)).ToList(), Who, Ct))
.ShouldAllBe(code => code == HistorianStatusCodes.GoodEntryInserted);
(await DaysAsync(c, c.SampleTable)).ShouldBe(older.Select(t => t.Date));
(await DaysAsync(c, c.SampleModifiedTable)).ShouldBe(older.Select(t => t.Date));
// The far-future row, then one more ordinary commit (TTL housekeeping runs on commits).
(await updater.UpdateDataAsync("ttl.tag", HistoryUpdateMode.Insert,
[V(new DateTime(2066, 1, 1, 0, 0, 0, DateTimeKind.Utc), 9)], Who, Ct)).ShouldBe([HistorianStatusCodes.GoodEntryInserted]);
(await updater.UpdateDataAsync("ttl.tag", HistoryUpdateMode.Insert, [V(today, 10)], Who, Ct))
.ShouldBe([HistorianStatusCodes.GoodEntryInserted]);
(await wal.WaitAsync(c.SampleTable, Ct)).ShouldBeTrue();
(await wal.WaitAsync(c.SampleModifiedTable, Ct)).ShouldBeTrue();
// Only the genuinely expired 40-day partition is gone; the 20- and 10-day partitions survive in both tables.
DateTime[] expected = [today.AddDays(-20).Date, today.AddDays(-10).Date, today.Date, new(2066, 1, 1)];
(await DaysAsync(c, c.SampleTable)).ShouldBe(expected);
(await DaysAsync(c, c.SampleModifiedTable)).ShouldBe(expected);
}
private static HistorizationValue V(DateTime t, double v) => new(t, HistorianValue.FromDouble(v), HistorianStatusCodes.Good);
/// <summary>The distinct UTC days holding rows, ascending.</summary>
private static async Task<List<DateTime>> DaysAsync(QuestDbConnection c, string table)
{
await using var conn = await c.OpenAsync(Ct);
await using var cmd = c.CreateCommand(conn, $"SELECT ts FROM {table} ORDER BY ts");
await using var r = await cmd.ExecuteReaderAsync(Ct);
var days = new SortedSet<DateTime>();
while (await r.ReadAsync(Ct)) days.Add(QuestDbConnection.ReadUtc(r, 0).Date);
return days.ToList();
}
}
@@ -14,6 +14,11 @@ public sealed class StatusCodeOracleTests
((uint)Opc.Ua.AggregateBits.Calculated).ShouldBe(HistorianInfoBits.HistorianCalculated);
}
/// <summary>#674: the v1-schema refusal code is the SDK's BadHistoryOperationUnsupported.</summary>
[Fact]
public void History_operation_unsupported_matches_the_SDK() =>
HistorianStatusCodes.BadHistoryOperationUnsupported.ShouldBe(Opc.Ua.StatusCodes.BadHistoryOperationUnsupported);
[Fact]
public void Interpolated_bits_match_the_SDK()
{
@@ -54,7 +54,7 @@ public sealed class ModifiedHistoryAtomicityTests(SqlHistorianFixture fx)
public async Task A_replace_whose_trail_insert_fails_changes_nothing()
{
var db = await ReadTestDb.CreateAsync(fx);
var u = new SqlServerHistorianDataUpdater(db.Conn, new SqlServerTagCatalog(db.Conn, db.Boot));
var u = new SqlServerHistorianDataUpdater(db.Conn, db.Boot, new SqlServerTagCatalog(db.Conn, db.Boot));
(await u.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T1, 1)], Who, CancellationToken.None))
.ShouldBe([HistorianStatusCodes.GoodEntryInserted]);
(await CountTrailAsync(db)).ShouldBe(1);
@@ -78,7 +78,7 @@ public sealed class ModifiedHistoryAtomicityTests(SqlHistorianFixture fx)
public async Task Deletes_whose_trail_insert_fails_delete_nothing()
{
var db = await ReadTestDb.CreateAsync(fx);
var u = new SqlServerHistorianDataUpdater(db.Conn, new SqlServerTagCatalog(db.Conn, db.Boot));
var u = new SqlServerHistorianDataUpdater(db.Conn, db.Boot, new SqlServerTagCatalog(db.Conn, db.Boot));
await u.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T1, 1), V(T2, 2)], Who, CancellationToken.None);
await BreakTheTrailAsync(db);
@@ -99,7 +99,7 @@ public sealed class ModifiedHistoryAtomicityTests(SqlHistorianFixture fx)
public async Task An_over_long_principal_is_truncated_never_refused()
{
var db = await ReadTestDb.CreateAsync(fx);
var u = new SqlServerHistorianDataUpdater(db.Conn, new SqlServerTagCatalog(db.Conn, db.Boot));
var u = new SqlServerHistorianDataUpdater(db.Conn, db.Boot, new SqlServerTagCatalog(db.Conn, db.Boot));
var who = new HistoryUpdateContext(new string('p', 500), new string('k', 500));
(await u.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T1, 1)], who, CancellationToken.None))
@@ -178,6 +178,97 @@ public sealed class RetentionTests(SqlHistorianFixture fx)
(await LockIsFreeAsync(s)).ShouldBeTrue();
}
[Fact]
public async Task Cancelling_mid_command_while_the_delete_is_blocked_still_releases_the_applock()
{
fx.SkipIfUnavailable();
var (c, b, _, s) = Make();
await SeedAsync(b, s);
// A second session holds an exclusive table lock, so the purger's first DELETE blocks server-side.
await using var holder = new SqlConnection(fx.ConnectionString);
await holder.OpenAsync();
await using var tx = (SqlTransaction)await holder.BeginTransactionAsync();
await using (var cmd = holder.CreateCommand())
{
cmd.Transaction = tx;
cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Sample WITH (TABLOCKX, HOLDLOCK)";
await cmd.ExecuteScalarAsync();
}
short holderSpid;
await using (var cmd = holder.CreateCommand())
{
cmd.Transaction = tx;
cmd.CommandText = "SELECT CAST(@@SPID AS smallint)";
holderSpid = (short)(await cmd.ExecuteScalarAsync())!;
}
using var cts = new CancellationTokenSource();
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now));
var purge = purger.PurgeAsync(cts.Token);
// Cancel only once the server reports a DELETE actually blocked behind the holder — a fixed delay could
// cancel during bootstrap/applock acquisition on a slow host and never exercise the mid-command path.
var deadline = DateTime.UtcNow.AddSeconds(15);
while (await CountAsync(
$"SELECT COUNT(*) FROM sys.dm_exec_requests WHERE blocking_session_id = {holderSpid} AND command = 'DELETE'") == 0)
{
purge.IsCompleted.ShouldBeFalse("the purge finished without ever blocking on the holder's table lock");
DateTime.UtcNow.ShouldBeLessThan(deadline, "the purge's DELETE never blocked behind the holder");
await Task.Delay(50);
}
cts.Cancel();
// SqlClient surfaces the attention as OperationCanceledException or as a SqlException.
var ex = await Should.ThrowAsync<Exception>(() => purge);
(ex is OperationCanceledException or SqlException).ShouldBeTrue($"unexpected {ex.GetType().Name}: {ex.Message}");
await tx.RollbackAsync();
(await LockIsFreeAsync(s)).ShouldBeTrue();
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(21);
}
[Fact]
public async Task A_failing_release_falls_back_to_clearing_the_pool_so_the_applock_is_freed()
{
fx.SkipIfUnavailable();
var (c, b, _, s) = Make();
await SeedAsync(b, s);
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now))
{
ReleaseSql = "EXEC sp_this_procedure_does_not_exist;",
};
await purger.PurgeAsync(CancellationToken.None);
(await LockIsFreeAsync(s)).ShouldBeTrue();
}
[Fact]
public async Task Falsification_control_a_failing_release_with_a_no_op_ClearPool_leaves_the_applock_held()
{
fx.SkipIfUnavailable();
var (c, b, _, s) = Make();
await SeedAsync(b, s);
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now))
{
ReleaseSql = "EXEC sp_this_procedure_does_not_exist;",
ClearPool = _ => { },
};
await purger.PurgeAsync(CancellationToken.None);
try
{
// The pooled session still owns the Session-scoped lock: proves the fallback test above is not vacuous.
(await LockIsFreeAsync(s)).ShouldBeFalse();
}
finally
{
SqlConnection.ClearAllPools(); // drop the leaked session so the lock does not outlive the test
}
}
[Fact]
public async Task A_throwing_purge_releases_the_applock()
{
@@ -32,7 +32,7 @@ public sealed class SchemaBootstrapIntegrationTests(SqlHistorianFixture fx)
var s = SqlHistorianFixture.UniqueSchema();
await Boot(Opts(s)).EnsureAsync(CancellationToken.None);
await Boot(Opts(s)).EnsureAsync(CancellationToken.None); // fresh instance => re-runs the guarded path
(await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(SqlServerSchemaBootstrapper.RequiredVersion);
(await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(SqlServerSchemaBootstrapper.ModifiedTrailVersion);
(await ScalarAsync($"SELECT COUNT(*) FROM sys.indexes WHERE name='CX_Event_Time' AND object_id=OBJECT_ID('[{s}].Event')")).ShouldBe(1);
}
@@ -51,13 +51,14 @@ public sealed class SchemaBootstrapIntegrationTests(SqlHistorianFixture fx)
cmd.CommandText = b;
await cmd.ExecuteNonQueryAsync();
}
(await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(SqlServerSchemaBootstrapper.RequiredVersion);
(await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(SqlServerSchemaBootstrapper.ModifiedTrailVersion);
}
/// <summary>
/// #668 upgrade path: a schema left at version 1 (only <c>001_initial.sql</c> applied, as every pre-#668 install is)
/// is upgraded in place by an auto-creating bootstrap — <c>SampleModified</c> appears, both version rows exist, and
/// existing samples are untouched. With <c>AutoCreateSchema=false</c> the same v1 schema is refused.
/// existing samples are untouched. With <c>AutoCreateSchema=false</c> the same v1 schema is accepted with the
/// modification trail switched off and left at v1 (#674).
/// </summary>
[Fact]
public async Task A_version_1_schema_is_upgraded_to_version_2_in_place()
@@ -81,11 +82,15 @@ public sealed class SchemaBootstrapIntegrationTests(SqlHistorianFixture fx)
}
(await ScalarAsync($"SELECT COUNT(*) FROM sys.tables WHERE name = 'SampleModified' AND schema_id = SCHEMA_ID('{s}')")).ShouldBe(0);
var refused = await Should.ThrowAsync<HistorianSchemaMissingException>(
() => Boot(Opts(s, auto: false)).EnsureAsync(CancellationToken.None));
refused.Message.ShouldContain("missing version 2 (found 1)");
// #674: verify-only accepts v1 — with the modification trail switched off — and does not upgrade it.
var verifyOnly = Boot(Opts(s, auto: false));
await verifyOnly.EnsureAsync(CancellationToken.None);
verifyOnly.SupportsModifiedTrail.ShouldBeFalse();
(await ScalarAsync($"SELECT MAX(Version) FROM [{s}].SchemaVersion")).ShouldBe(1);
await Boot(Opts(s)).EnsureAsync(CancellationToken.None);
var upgrading = Boot(Opts(s));
await upgrading.EnsureAsync(CancellationToken.None);
upgrading.SupportsModifiedTrail.ShouldBeTrue();
(await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(2);
(await ScalarAsync($"SELECT MAX(Version) FROM [{s}].SchemaVersion")).ShouldBe(2);
@@ -101,7 +106,7 @@ public sealed class SchemaBootstrapIntegrationTests(SqlHistorianFixture fx)
var s = SqlHistorianFixture.UniqueSchema();
var tasks = Enumerable.Range(0, 4).Select(_ => Task.Run(() => Boot(Opts(s)).EnsureAsync(CancellationToken.None))).ToArray();
await Task.WhenAll(tasks);
(await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(SqlServerSchemaBootstrapper.RequiredVersion);
(await ScalarAsync($"SELECT COUNT(*) FROM [{s}].SchemaVersion")).ShouldBe(SqlServerSchemaBootstrapper.ModifiedTrailVersion);
}
[Fact]
@@ -110,7 +115,7 @@ public sealed class SchemaBootstrapIntegrationTests(SqlHistorianFixture fx)
fx.SkipIfUnavailable();
var ex = await Should.ThrowAsync<HistorianSchemaMissingException>(
() => Boot(Opts(SqlHistorianFixture.UniqueSchema(), auto: false)).EnsureAsync(CancellationToken.None));
ex.Message.ShouldContain("missing version 2");
ex.Message.ShouldContain("missing version 1 (found 0)");
}
[Fact]
@@ -0,0 +1,181 @@
using Microsoft.Data.SqlClient;
using Microsoft.Extensions.Logging;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
/// <summary>
/// Gitea #674: a database left at schema version 1 (no Part 11 <c>SampleModified</c> trail) with
/// <c>AutoCreateSchema=false</c> keeps historizing. Before #674 the bootstrapper required v2 and refused EVERY call —
/// live writes, raw reads, alarm events, retention — until script 002 was applied by hand. Now only the features
/// that need the trail are refused (<c>BadHistoryOperationUnsupported</c>, nothing sent to SQL), and the
/// bootstrapper logs ONE warning for the whole provider. Built through <see cref="SqlServerHistorianProvider"/>, the
/// production wiring, so every seam shares the one bootstrapper the way a node does.
/// </summary>
[Collection(SqlHistorianCollection.Name)]
public sealed class SchemaV1PostureTests(SqlHistorianFixture fx)
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static readonly DateTime Recent = DateTime.UtcNow.AddHours(-1);
private static readonly HistoryUpdateContext Who = new("editor", "k1");
[Fact]
public async Task A_v1_schema_keeps_historizing_and_refuses_only_the_modified_trail_features()
{
fx.SkipIfUnavailable();
var s = SqlHistorianFixture.UniqueSchema();
await ApplyV1Async(s);
var old = DateTime.UtcNow.AddDays(-40);
var loggers = new CapturingLoggerFactory();
var provider = new SqlServerHistorianProvider(new SqlServerHistorianOptions
{
ConnectionString = fx.ConnectionString, Schema = s, AutoCreateSchema = false, RetentionDays = 30,
});
var sp = new LoggerServices(loggers);
// Live writes (the continuous-historization drain) and raw reads work.
var writer = provider.CreateValueWriter(sp);
(await writer.WriteLiveValuesAsync("Plant.Speed",
[new HistorizationValue(old, HistorianValue.FromDouble(1), 0), new HistorizationValue(Recent, HistorianValue.FromDouble(2), 0)],
Ct)).ShouldBeTrue();
var source = provider.CreateDataSource(sp);
var raw = await source.ReadRawAsync("Plant.Speed", Recent.AddMinutes(-1), Recent.AddMinutes(1), 10, Ct);
raw.Samples.Select(v => v.Value).ShouldBe([2d]);
// Alarm events are written.
var alarms = provider.CreateAlarmWriter(sp);
(await alarms.WriteBatchAsync(
[new AlarmHistorianEvent("alarm-1", "Plant/Line/Eq", "HiTemp", "LimitAlarm", AlarmSeverity.High, "Activated", "m", "u", null, Recent)],
Ct)).ShouldBe([HistorianWriteOutcome.Ack]);
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Event")).ShouldBe(1);
// Modified reads: the interface's "no trail" signal (the node manager reports BadHistoryOperationUnsupported),
// not a backend failure — health stays clean.
await Should.ThrowAsync<NotSupportedException>(() =>
source.ReadModifiedAsync("Plant.Speed", Recent.AddDays(-1), Recent.AddDays(1), 10, null, Ct));
source.GetHealthSnapshot().LastError.ShouldBeNull();
// HistoryUpdate: every item refused, and NOTHING reaches SQL — a sent MERGE/DELETE would either change the
// samples or fail on the missing SampleModified target; an Insert on an unknown tag would create its Tag row.
var updater = provider.CreateUpdater(sp)!;
var tagsBefore = await CountAsync($"SELECT COUNT(*) FROM [{s}].Tag");
(await updater.UpdateDataAsync("Plant.Speed", HistoryUpdateMode.Update,
[new HistorizationValue(Recent, HistorianValue.FromDouble(99), 0), new HistorizationValue(Recent.AddSeconds(1), HistorianValue.FromDouble(98), 0)],
Who, Ct)).ShouldBe([HistorianStatusCodes.BadHistoryOperationUnsupported, HistorianStatusCodes.BadHistoryOperationUnsupported]);
(await updater.UpdateDataAsync("Plant.Brand.New", HistoryUpdateMode.Insert,
[new HistorizationValue(Recent, HistorianValue.FromDouble(1), 0)], Who, Ct))
.ShouldBe([HistorianStatusCodes.BadHistoryOperationUnsupported]);
(await updater.DeleteRawAsync("Plant.Speed", Recent.AddMinutes(-1), Recent.AddMinutes(1), Who, Ct))
.ShouldBe(HistorianStatusCodes.BadHistoryOperationUnsupported);
(await updater.DeleteAtTimeAsync("Plant.Speed", [Recent], Who, Ct))
.ShouldBe([HistorianStatusCodes.BadHistoryOperationUnsupported]);
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Tag")).ShouldBe(tagsBefore);
(await source.ReadRawAsync("Plant.Speed", Recent.AddMinutes(-1), Recent.AddMinutes(1), 10, Ct))
.Samples.Select(v => v.Value).ShouldBe([2d]);
// Retention purges samples and skips the trail it does not have.
var purge = await provider.CreateRetention(sp)!.PurgeAsync(Ct);
purge.Select(r => r.Table).ShouldBe(["Sample", "Event"]);
purge.Single(r => r.Table == "Sample").Rows.ShouldBe(1);
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(1);
// One warning for the whole provider (= per boot), however many seams and calls touched the bootstrapper.
loggers.Warnings.Count(w => w.Contains("Part 11 modification trail is unavailable")).ShouldBe(1);
(await CountAsync($"SELECT MAX(Version) FROM [{s}].SchemaVersion")).ShouldBe(1); // verify-only: never upgraded
}
/// <summary>
/// #674: <c>AutoCreateSchema=true</c> still upgrades a v1 schema to v2 — the modification trail appears and a
/// HistoryUpdate is recorded in it — and logs no v1 warning.
/// </summary>
[Fact]
public async Task Auto_create_still_upgrades_a_v1_schema_and_enables_the_trail()
{
fx.SkipIfUnavailable();
var s = SqlHistorianFixture.UniqueSchema();
await ApplyV1Async(s);
var loggers = new CapturingLoggerFactory();
var provider = new SqlServerHistorianProvider(new SqlServerHistorianOptions
{
ConnectionString = fx.ConnectionString, Schema = s, AutoCreateSchema = true,
});
var sp = new LoggerServices(loggers);
(await provider.CreateUpdater(sp)!.UpdateDataAsync("Plant.Speed", HistoryUpdateMode.Insert,
[new HistorizationValue(Recent, HistorianValue.FromDouble(5), 0)], Who, Ct))
.ShouldBe([HistorianStatusCodes.GoodEntryInserted]);
(await CountAsync($"SELECT MAX(Version) FROM [{s}].SchemaVersion")).ShouldBe(SqlServerSchemaBootstrapper.ModifiedTrailVersion);
var trail = await provider.CreateDataSource(sp)
.ReadModifiedAsync("Plant.Speed", Recent.AddMinutes(-1), Recent.AddMinutes(1), 10, null, Ct);
trail.Values.ShouldHaveSingleItem().UpdateType.ShouldBe(HistoryModificationType.Insert);
loggers.Warnings.ShouldNotContain(w => w.Contains("Part 11 modification trail is unavailable"));
}
private async Task ApplyV1Async(string schema)
{
var v1 = new StreamReader(typeof(SqlServerSchemaBootstrapper).Assembly.GetManifestResourceStream("Schema.001_initial.sql")!).ReadToEnd();
await using var c = new SqlConnection(fx.ConnectionString);
await c.OpenAsync(Ct);
foreach (var b in SqlServerSchemaBootstrapper.SplitBatches(v1.Replace("$(Schema)", schema)))
{
await using var cmd = c.CreateCommand();
cmd.CommandText = b;
await cmd.ExecuteNonQueryAsync(Ct);
}
}
private async Task<long> CountAsync(string sql)
{
await using var c = new SqlConnection(fx.ConnectionString);
await c.OpenAsync(Ct);
await using var cmd = c.CreateCommand();
cmd.CommandText = sql;
return Convert.ToInt64(await cmd.ExecuteScalarAsync(Ct));
}
/// <summary>The only service the provider resolves is <see cref="ILoggerFactory"/>.</summary>
private sealed class LoggerServices(ILoggerFactory loggers) : IServiceProvider
{
public object? GetService(Type serviceType) => serviceType == typeof(ILoggerFactory) ? loggers : null;
}
/// <summary>Captures every Warning-or-above message from every logger it creates.</summary>
private sealed class CapturingLoggerFactory : ILoggerFactory
{
private readonly List<string> _warnings = [];
public IReadOnlyList<string> Warnings
{
get { lock (_warnings) return _warnings.ToList(); }
}
public ILogger CreateLogger(string categoryName) => new Capturing(this);
public void AddProvider(ILoggerProvider provider)
{
}
public void Dispose()
{
}
private sealed class Capturing(CapturingLoggerFactory owner) : ILogger
{
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) return;
lock (owner._warnings) owner._warnings.Add(formatter(state, exception));
}
}
}
}
@@ -60,3 +60,10 @@ public sealed class SqlServerModifiedValueConformance(SqlHistorianFixture fx) :
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => SqlServerConformanceBackend.CreateAsync(fx);
}
[Collection(SqlHistorianCollection.Name)]
public sealed class SqlServerEventReadConformance(SqlHistorianFixture fx) : EventReadConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => SqlServerConformanceBackend.CreateAsync(fx);
}
@@ -34,12 +34,28 @@ internal sealed class SqlServerConformanceBackend : IConformanceBackend
/// <inheritdoc/>
public IHistorianDataUpdater CreateUpdater() =>
new SqlServerHistorianDataUpdater(_db.Conn, new SqlServerTagCatalog(_db.Conn, _db.Boot));
new SqlServerHistorianDataUpdater(_db.Conn, _db.Boot, new SqlServerTagCatalog(_db.Conn, _db.Boot));
/// <inheritdoc/>
public async Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples) =>
await _db.SeedAsync(tag, samples);
/// <inheritdoc/>
public async Task SeedEventsAsync(params (DateTime T, string Id, string Source)[] events)
{
await using var c = await _db.Conn.OpenAsync(CancellationToken.None);
foreach (var (t, id, source) in events)
{
await using var cmd = _db.Conn.CreateCommand(c,
$"INSERT {_db.Conn.Q("Event")} (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, " +
"UserName, Comment, EventTimeUtc, ReceivedTimeUtc) VALUES (@id, @src, 'a', 'n', 'T', 500, 'Activated', NULL, NULL, NULL, @t, @t)");
cmd.Parameters.AddWithValue("@id", id);
cmd.Parameters.AddWithValue("@src", source);
SqlHistorianConnection.AddDateTime2(cmd, "@t", t);
await cmd.ExecuteNonQueryAsync();
}
}
/// <inheritdoc/>
public async Task<IReadOnlyList<(DateTime T, HistorianValue V, uint Status)>> ReadBackAsync(string tag)
{
@@ -32,7 +32,7 @@ public sealed class SchemaScriptTests
/// <summary>
/// #668: schema v2 (the Part 11 modification trail) is embedded, matches its docs copy, records version 2 (which
/// must equal <see cref="SqlServerSchemaBootstrapper.RequiredVersion"/>), and carries none of the constructs an
/// must equal <see cref="SqlServerSchemaBootstrapper.ModifiedTrailVersion"/>), and carries none of the constructs an
/// <c>OUTPUT … INTO</c> target forbids (FOREIGN KEY / CHECK / trigger) — any of them would fail every HistoryUpdate.
/// </summary>
[Fact]
@@ -45,7 +45,8 @@ public sealed class SchemaScriptTests
text.ShouldContain("[SampleModified]");
text.ShouldContain("PRIMARY KEY CLUSTERED (TagId, SourceTimeUtc, ModifiedId)");
text.ShouldContain("VALUES (2, SYSUTCDATETIME())");
SqlServerSchemaBootstrapper.RequiredVersion.ShouldBe(2);
SqlServerSchemaBootstrapper.ModifiedTrailVersion.ShouldBe(2);
SqlServerSchemaBootstrapper.RequiredVersion.ShouldBe(1); // #674: v1 is still served, without the trail
var ddl = string.Join('\n', text.Split('\n').Where(l => !l.TrimStart().StartsWith("--", StringComparison.Ordinal)));
ddl.ShouldNotContain("FOREIGN KEY", Case.Insensitive);
ddl.ShouldNotContain("CHECK", Case.Insensitive);