Merge feat/674-historyupdate-posture: future-skew bound, Primary-only HistoryUpdate, SqlServer v1 accepted with modified trail disabled, QuestDB TTL pin (#674)
This commit is contained in:
+3
-3
@@ -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))
|
||||
|
||||
+14
-9
@@ -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]
|
||||
|
||||
+181
@@ -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));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -34,7 +34,7 @@ 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) =>
|
||||
|
||||
Reference in New Issue
Block a user