diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs new file mode 100644 index 00000000..c818e8b0 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs @@ -0,0 +1,258 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Npgsql; +using NpgsqlTypes; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; + +/// +/// OPC UA HistoryUpdate against the QuestDB sample table (spec §2.3). Every operation is an INSERT: the +/// table's DEDUP UPSERT KEYS(ts, tag) performs the replace, and a delete writes a tombstone +/// (deleted = true, NULL values, BadNoData). QuestDB UPDATE/DELETE are never issued. +/// Every status comes from , so codes equal the SQL Server backend's. +/// "Exists" means present and not deleted, so Insert over a tombstone succeeds (GoodEntryInserted) while +/// Replace and DeleteAtTime treat a tombstone as no entry. Existence is read with one range query and filtered +/// client-side (never an IN list). +/// Pre-1970 (RC1): QuestDB cannot store such a designated timestamp. Insert/Replace/Update of such an +/// item reports BadOutOfRange and sends nothing (a null/MinValue timestamp is still +/// BadInvalidTimestamp); DeleteAtTime reports BadNoEntryExists; a DeleteRaw window may start before +/// 1970 (read bounds are not clamped). +/// Read-your-writes: after writing, waits for the rows to become +/// visible, bounded by VisibilityTimeout; a timeout only logs a Warning, never changes a code. +/// +internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater +{ + /// Rows per INSERT and per DeleteRaw page (T0 Q2: 9,000 parameters, far under the 32,767 limit). + internal const int BatchSize = 1000; + + private const long ResolutionTicks = 10; + private readonly QuestDbConnection _conn; + private readonly QuestDbTagCatalog _tags; + private readonly QuestDbWalVisibility _wal; + private readonly ILogger _log; + + /// Creates the updater. + /// Connection helper. + /// Tag catalog (R7: the only known-tag authority). + /// WAL visibility wait (R5). + /// Optional logger. + public QuestDbHistorianDataUpdater(QuestDbConnection conn, QuestDbTagCatalog tags, QuestDbWalVisibility wal, + ILogger? log = null) + { + _conn = conn; + _tags = tags; + _wal = wal; + _log = log ?? NullLogger.Instance; + } + + /// + public async Task> UpdateDataAsync(string tag, HistoryUpdateMode mode, + IReadOnlyList values, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(tag); + ArgumentNullException.ThrowIfNull(values); + + // RC1: a valid-timestamp item before 1970 can never be stored; refuse it and plan the rest. + var final = new uint[values.Count]; + var kept = new List(values.Count); + for (var i = 0; i < values.Count; i++) + { + if (values[i].TimestampUtc is { } ts && ts != DateTime.MinValue && !QuestDbConnection.IsStorable(ts)) + final[i] = HistorianStatusCodes.BadOutOfRange; + else + kept.Add(i); + } + + var plan = HistoryUpdatePlanner.Plan(tag, mode, kept.Select(i => values[i]).ToList(), ResolutionTicks); + IReadOnlyList planned; + if (plan.Winners.Count == 0) + { + planned = HistoryUpdatePlanner.Complete(plan, mode, new Dictionary()); + } + else + { + var known = mode == HistoryUpdateMode.Replace + ? await _tags.ExistsAsync(tag, ct).ConfigureAwait(false) + : await EnsureAsync(tag, plan.Winners[0].Row.Value.Kind, ct).ConfigureAwait(false); + var outcome = known + ? await ApplyAsync(tag, mode, plan.Winners.Select(w => w.Row).ToList(), ct).ConfigureAwait(false) + : new Dictionary(); + planned = HistoryUpdatePlanner.Complete(plan, mode, outcome); + } + for (var k = 0; k < kept.Count; k++) final[kept[k]] = planned[k]; + return final; + } + + /// + public async Task DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(tag); + if (!HistorianTagNames.IsValid(tag)) return HistorianStatusCodes.BadInvalidArgument; + if (!await _tags.ExistsAsync(tag, ct).ConfigureAwait(false)) + return HistoryUpdatePlanner.DeleteRawResult(tagKnown: false, 0); + + var cursor = QuestDbConnection.Round(startUtc); + var end = QuestDbConnection.Round(endUtc); + long total = 0; + while (cursor < end) + { + var page = new List(BatchSize); + await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false)) + await using (var cmd = _conn.CreateCommand(conn, + $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts < $3 AND NOT deleted ORDER BY ts LIMIT {BatchSize}")) + { + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindTimestamp(cmd, cursor); + QuestDbConnection.BindTimestamp(cmd, end); + await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await rd.ReadAsync(ct).ConfigureAwait(false)) page.Add(QuestDbConnection.ReadUtc(rd, 0)); + } + if (page.Count == 0) break; + await WriteTombstonesAsync(tag, page, ct).ConfigureAwait(false); + // The wait between pages is required: without it the next page could re-read rows just tombstoned. + await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); + total += page.Count; + if (page.Count < BatchSize) break; + cursor = page[^1].AddTicks(ResolutionTicks); + } + return HistoryUpdatePlanner.DeleteRawResult(tagKnown: true, total); + } + + /// + public async Task> DeleteAtTimeAsync(string tag, IReadOnlyList timesUtc, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(tag); + ArgumentNullException.ThrowIfNull(timesUtc); + if (!HistorianTagNames.IsValid(tag)) + { + var refused = new uint[timesUtc.Count]; + Array.Fill(refused, HistorianStatusCodes.BadInvalidArgument); + return refused; + } + + // M-2: round BEFORE both the delete and the comparison, so a sub-µs request time matches the row it deleted. + // MinValue stays as is: the planner turns it into BadInvalidTimestamp. + var rounded = timesUtc.Select(t => t == DateTime.MinValue ? t : QuestDbConnection.Round(t)).ToList(); + var candidates = new HashSet(); + foreach (var t in rounded) + if (t != DateTime.MinValue && t >= DateTime.UnixEpoch) candidates.Add(t.Ticks); // pre-1970 can hold no row + var deleted = new HashSet(); + if (candidates.Count == 0 || !await _tags.ExistsAsync(tag, ct).ConfigureAwait(false)) + return HistoryUpdatePlanner.DeleteAtTimeResults(rounded, deleted, tagKnown: false); + + var live = await ReadLiveAsync(tag, candidates.Min(), candidates.Max(), ct).ConfigureAwait(false); + var toDelete = candidates.Where(live.Contains).OrderBy(x => x).ToList(); + foreach (var chunk in toDelete.Chunk(BatchSize)) + await WriteTombstonesAsync(tag, chunk.Select(t => new DateTime(t, DateTimeKind.Utc)).ToList(), ct).ConfigureAwait(false); + if (toDelete.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); + deleted.UnionWith(toDelete); + return HistoryUpdatePlanner.DeleteAtTimeResults(rounded, deleted, tagKnown: true); + } + + private async Task EnsureAsync(string tag, HistorianValueKind kind, CancellationToken ct) + { + await _tags.EnsureAsync(tag, HistorianDataTypes.FromKind(kind), ct).ConfigureAwait(false); + return true; + } + + private async Task> ApplyAsync(string tag, HistoryUpdateMode mode, + List rows, CancellationToken ct) + { + // Key = the planner's tick (what Complete looks up); the stored time is Round(of the same row). + var live = await ReadLiveAsync(tag, rows.Min(r => QuestDbConnection.Round(r.TimeUtc).Ticks), + rows.Max(r => QuestDbConnection.Round(r.TimeUtc).Ticks), ct).ConfigureAwait(false); + var outcome = new Dictionary(); + var toWrite = new List(); + foreach (var r in rows) + { + var present = live.Contains(QuestDbConnection.Round(r.TimeUtc).Ticks); + if (mode == HistoryUpdateMode.Insert && present) continue; + if (mode == HistoryUpdateMode.Replace && !present) continue; + outcome[r.TimeUtc.Ticks] = present ? UpsertOutcome.Replaced : UpsertOutcome.Inserted; + toWrite.Add(r); + } + foreach (var chunk in toWrite.Chunk(BatchSize)) + await WriteRowsAsync(tag, chunk, ct).ConfigureAwait(false); + if (toWrite.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); + return outcome; + } + + /// Ticks of the live (present, not deleted) rows with ts in [minTicks, maxTicks]: one range query. + private async Task> ReadLiveAsync(string tag, long minTicks, long maxTicks, CancellationToken ct) + { + var live = new HashSet(); + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, + $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3 AND NOT deleted"); + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindTimestamp(cmd, new DateTime(minTicks, DateTimeKind.Utc)); + QuestDbConnection.BindTimestamp(cmd, new DateTime(maxTicks, DateTimeKind.Utc)); + await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await rd.ReadAsync(ct).ConfigureAwait(false)) live.Add(QuestDbConnection.ReadUtc(rd, 0).Ticks); + return live; + } + + private Task WriteRowsAsync(string tag, IReadOnlyList rows, CancellationToken ct) => + InsertAsync(rows.Count, ct, (cmd, i) => + { + var row = rows[i]; + var c = SampleColumnMapper.ToColumns(row.Value); + QuestDbConnection.BindTimestamp(cmd, row.TimeUtc, clampToEpoch: true); + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindByte(cmd, c.ValueKind); + Typed(cmd, NpgsqlDbType.Double, c.NumValue); + Typed(cmd, NpgsqlDbType.Bigint, c.IntValue); + Typed(cmd, NpgsqlDbType.Varchar, c.StrValue); + QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(row.StatusCode)); + Typed(cmd, NpgsqlDbType.Boolean, false); + }); + + private Task WriteTombstonesAsync(string tag, IReadOnlyList times, CancellationToken ct) => + InsertAsync(times.Count, ct, (cmd, i) => + { + QuestDbConnection.BindTimestamp(cmd, times[i], clampToEpoch: true); + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Empty); + Typed(cmd, NpgsqlDbType.Double, null); + Typed(cmd, NpgsqlDbType.Bigint, null); + Typed(cmd, NpgsqlDbType.Varchar, null); + QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(HistorianStatusCodes.BadNoData)); + Typed(cmd, NpgsqlDbType.Boolean, true); + }); + + /// Multi-row INSERT; adds the first eight parameters of a row, this adds server_ts. + private async Task InsertAsync(int count, CancellationToken ct, Action bindRow) + { + if (count == 0) return; + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, ""); + var now = DateTime.UtcNow; + var sb = new System.Text.StringBuilder( + $"INSERT INTO {_conn.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES "); + for (var i = 0; i < count; i++) + { + bindRow(cmd, i); + QuestDbConnection.BindTimestamp(cmd, now); + var b = i * 9; + if (i > 0) sb.Append(','); + sb.Append('('); + for (var p = 1; p <= 9; p++) sb.Append(p > 1 ? "," : "").Append('$').Append(b + p); + sb.Append(')'); + } + cmd.CommandText = sb.ToString(); + try + { + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _log.LogWarning(ex, "QuestDB HistoryUpdate INSERT of {Count} rows into {Table} failed.", count, _conn.SampleTable); + throw; + } + } + + private static void Typed(NpgsqlCommand cmd, NpgsqlDbType type, object? value) => + cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = type, Value = value ?? DBNull.Value }); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs new file mode 100644 index 00000000..2f3f53e3 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs @@ -0,0 +1,50 @@ +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; + +/// Soft-delete behaviour. The writer-redelivery case lives in T7 (TombstoneTests.Redelivery.cs). +[Collection(QuestDbCollection.Name)] +public sealed class TombstoneTests(QuestDbFixture fx) +{ + private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc); + + [Fact] + public async Task Delete_then_raw_read_hides_the_row_and_leaves_a_NULL_valued_tombstone() + { + fx.SkipIfUnavailable(); + var rig = UpdateTests.MakeRig(fx); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, + [UpdateTests.V(T0, 1), UpdateTests.V(T0.AddSeconds(1), 2)], default); + (await rig.Updater.DeleteAtTimeAsync("t", [T0], default)).ShouldBe([HistorianStatusCodes.Good]); + + (await UpdateTests.RowsAsync(rig, "t", liveOnly: true)).Select(r => r.Num).ShouldBe([2d]); + var all = await UpdateTests.RowsAsync(rig, "t"); + all.Count.ShouldBe(2); // the tombstone replaced the row in place (dedup upsert) + var tomb = all[0]; + tomb.Deleted.ShouldBeTrue(); + tomb.Num.ShouldBeNull(); + tomb.Kind.ShouldBe((byte)0); + tomb.Status.ShouldBe((long)HistorianStatusCodes.BadNoData); + } + + [Fact] + public async Task Insert_over_a_tombstone_reports_GoodEntryInserted_and_the_row_is_visible() + { + fx.SkipIfUnavailable(); + var rig = UpdateTests.MakeRig(fx); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], default); + await rig.Updater.DeleteAtTimeAsync("t", [T0], default); + + (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 5)], default)) + .ShouldBe([HistorianStatusCodes.BadNoEntryExists]); // a tombstone is no entry for Replace + (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 9)], default)) + .ShouldBe([HistorianStatusCodes.GoodEntryInserted]); + var live = await UpdateTests.RowsAsync(rig, "t", liveOnly: true); + live.Count.ShouldBe(1); + live[0].Num.ShouldBe(9d); + live[0].Deleted.ShouldBeFalse(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs new file mode 100644 index 00000000..c77f3d64 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs @@ -0,0 +1,235 @@ +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; + +/// QuestDB-specific checks of (the literal codes run through conformance in T7). +[Collection(QuestDbCollection.Name)] +public sealed class UpdateTests(QuestDbFixture fx) +{ + private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc); // µs-aligned + private const uint Good = HistorianStatusCodes.Good; + + internal sealed record Rig(QuestDbConnection Conn, QuestDbHistorianDataUpdater Updater, QuestDbTagCatalog Tags); + + internal static Rig MakeRig(QuestDbFixture fx, TimeSpan? visibility = null) + { + var o = fx.Options(fx.UniquePrefix()); + o.VisibilityTimeout = visibility ?? o.VisibilityTimeout; + var c = new QuestDbConnection(o); + var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout); + var tags = new QuestDbTagCatalog(c, new QuestDbSchemaBootstrapper(c, o), wal); + return new Rig(c, new QuestDbHistorianDataUpdater(c, tags, wal), tags); + } + + internal static HistorizationValue V(DateTime? t, double v, uint status = Good) => + new(t, HistorianValue.FromDouble(v), status); + + internal sealed record Row(DateTime Ts, byte Kind, double? Num, long? Status, bool Deleted); + + /// Every stored row for the tag (tombstones included), straight from the table. + internal static async Task> RowsAsync(Rig rig, string tag, bool liveOnly = false) + { + await using var conn = await rig.Conn.OpenAsync(default); + await using var cmd = rig.Conn.CreateCommand(conn, + $"SELECT ts, kind, num_value, status, deleted FROM {rig.Conn.SampleTable} WHERE tag = $1 {(liveOnly ? "AND NOT deleted" : "")} ORDER BY ts"); + QuestDbConnection.BindText(cmd, tag); + var rows = new List(); + await using var r = await cmd.ExecuteReaderAsync(); + while (await r.ReadAsync()) + rows.Add(new Row(QuestDbConnection.ReadUtc(r, 0), QuestDbConnection.ReadKind(r, 1), + r.IsDBNull(2) ? null : r.GetDouble(2), r.IsDBNull(3) ? null : r.GetInt64(3), r.GetBoolean(4))); + return rows; + } + + internal static async Task TagRowsAsync(Rig rig, string tag) + { + await using var conn = await rig.Conn.OpenAsync(default); + await using var cmd = rig.Conn.CreateCommand(conn, $"SELECT count() FROM {rig.Conn.TagTable} WHERE name = $1"); + QuestDbConnection.BindText(cmd, tag); + return Convert.ToInt64(await cmd.ExecuteScalarAsync()); + } + + [Fact] + public async Task Insert_writes_absent_and_refuses_existing_and_is_visible_immediately() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + (await rig.Updater.UpdateDataAsync("t1", HistoryUpdateMode.Insert, [V(T0, 1)], default)) + .ShouldBe([HistorianStatusCodes.GoodEntryInserted]); + var codes = await rig.Updater.UpdateDataAsync("t1", HistoryUpdateMode.Insert, + [V(T0, 99), V(T0.AddSeconds(1), 2)], default); + codes.ShouldBe([HistorianStatusCodes.BadEntryExists, HistorianStatusCodes.GoodEntryInserted]); + var rows = await RowsAsync(rig, "t1"); // read right after return: read-your-writes + rows.Select(r => r.Num).ShouldBe([1d, 2d]); + (await TagRowsAsync(rig, "t1")).ShouldBe(1); + } + + [Fact] + public async Task Replace_overwrites_present_refuses_missing_and_never_creates_the_tag() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + (await rig.Updater.UpdateDataAsync("nope", HistoryUpdateMode.Replace, [V(T0, 1)], default)) + .ShouldBe([HistorianStatusCodes.BadNoEntryExists]); + (await TagRowsAsync(rig, "nope")).ShouldBe(0); + (await RowsAsync(rig, "nope")).ShouldBeEmpty(); + + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1)], default); + var codes = await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, + [V(T0, 5), V(T0.AddSeconds(1), 6)], default); + codes.ShouldBe([HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.BadNoEntryExists]); + (await RowsAsync(rig, "t")).Select(r => r.Num).ShouldBe([5d]); + } + + [Fact] + public async Task Update_upserts_both_ways_and_last_duplicate_wins() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1)], default); + var codes = await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Update, + [V(T0, 2), V(T0.AddSeconds(1), 3), V(T0.AddSeconds(1).AddTicks(3), 4)], default); + codes.ShouldBe([HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.GoodEntryInserted, + HistorianStatusCodes.GoodEntryInserted]); + (await RowsAsync(rig, "t")).Select(r => r.Num).ShouldBe([2d, 4d]); + } + + [Fact] + public async Task Same_microsecond_duplicates_collapse_before_the_database_sees_them() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + // offsets 0 and 3 ticks sit inside one 10-tick window => the same stored µs + var codes = await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, + [V(T0, 1), V(T0.AddTicks(3), 2)], default); + codes.ShouldBe([HistorianStatusCodes.GoodEntryInserted, HistorianStatusCodes.BadEntryExists]); + (await RowsAsync(rig, "t")).Select(r => r.Num).ShouldBe([1d]); + } + + [Fact] + public async Task Invalid_items_get_the_planners_codes_and_the_rest_proceed() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + var codes = await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, + [ + V(null, 1), V(DateTime.MinValue, 1), V(T0, double.NaN), + new HistorizationValue(T0.AddSeconds(1), HistorianValue.Empty, Good), + V(T0.AddSeconds(2), 7), + ], default); + codes.ShouldBe([HistorianStatusCodes.BadInvalidTimestamp, HistorianStatusCodes.BadInvalidTimestamp, + HistorianStatusCodes.BadInvalidArgument, HistorianStatusCodes.BadInvalidArgument, + HistorianStatusCodes.GoodEntryInserted]); + (await RowsAsync(rig, "t")).Count.ShouldBe(1); + } + + [Fact] + public async Task Over_long_tag_is_refused_and_nothing_is_written() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + var tag = new string('x', HistorianTagNames.MaxLength + 1); + (await rig.Updater.UpdateDataAsync(tag, HistoryUpdateMode.Insert, [V(T0, 1)], default)) + .ShouldBe([HistorianStatusCodes.BadInvalidArgument]); + (await rig.Updater.DeleteAtTimeAsync(tag, [T0], default)).ShouldBe([HistorianStatusCodes.BadInvalidArgument]); + (await rig.Updater.DeleteRawAsync(tag, T0, T0.AddHours(1), default)).ShouldBe(HistorianStatusCodes.BadInvalidArgument); + } + + [Fact] + public async Task Pre_1970_items_are_BadOutOfRange_and_nothing_is_sent() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + var pre = new DateTime(1969, 12, 31, 23, 59, 59, DateTimeKind.Utc); + foreach (var mode in new[] { HistoryUpdateMode.Insert, HistoryUpdateMode.Replace, HistoryUpdateMode.Update }) + { + var codes = await rig.Updater.UpdateDataAsync("t", mode, [V(pre, 1), V(DateTime.MinValue, 1), V(T0, 2)], default); + codes[0].ShouldBe(HistorianStatusCodes.BadOutOfRange); + codes[1].ShouldBe(HistorianStatusCodes.BadInvalidTimestamp); + } + // Pre-1970 only: the whole batch is refused, no tag row is created, no INSERT would succeed anyway. + (await rig.Updater.UpdateDataAsync("only-pre", HistoryUpdateMode.Update, [V(pre, 1)], default)) + .ShouldBe([HistorianStatusCodes.BadOutOfRange]); + (await TagRowsAsync(rig, "only-pre")).ShouldBe(0); + (await rig.Updater.DeleteAtTimeAsync("t", [pre], default)).ShouldBe([HistorianStatusCodes.BadNoEntryExists]); + } + + [Fact] + public async Task DeleteAtTime_tombstones_live_rows_and_reports_per_time_codes() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1), V(T0.AddSeconds(1), 2)], default); + var codes = await rig.Updater.DeleteAtTimeAsync("t", + [T0, T0.AddSeconds(5), T0, DateTime.MinValue], default); + codes.ShouldBe([Good, HistorianStatusCodes.BadNoEntryExists, HistorianStatusCodes.BadNoEntryExists, + HistorianStatusCodes.BadInvalidTimestamp]); + (await RowsAsync(rig, "t", liveOnly: true)).Select(r => r.Num).ShouldBe([2d]); + // a tombstone is not a live row: deleting it again finds nothing + (await rig.Updater.DeleteAtTimeAsync("t", [T0], default)).ShouldBe([HistorianStatusCodes.BadNoEntryExists]); + (await rig.Updater.DeleteAtTimeAsync("unknown", [T0], default)).ShouldBe([HistorianStatusCodes.BadNoEntryExists]); + } + + [Fact] + public async Task DeleteAtTime_with_a_sub_microsecond_request_time_still_reports_Good_for_the_row_it_deleted() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1)], default); + // +7 ticks is inside the same µs; the row lives at T0. + (await rig.Updater.DeleteAtTimeAsync("t", [T0.AddTicks(7)], default)).ShouldBe([Good]); + (await RowsAsync(rig, "t", liveOnly: true)).ShouldBeEmpty(); + } + + [Fact] + public async Task DeleteRaw_is_half_open_pages_through_big_ranges_and_reports_BadNoData_when_empty() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + (await rig.Updater.DeleteRawAsync("unknown", T0, T0.AddHours(1), default)).ShouldBe(HistorianStatusCodes.BadNoData); + + // 2,500 rows => 3 pages of at most 1,000; plus one row exactly at the (exclusive) end. + var values = Enumerable.Range(0, 2500).Select(i => V(T0.AddSeconds(i), i)).ToList(); + values.Add(V(T0.AddSeconds(2500), -1)); + (await rig.Updater.UpdateDataAsync("big", HistoryUpdateMode.Insert, values, default)) + .ShouldAllBe(c => c == HistorianStatusCodes.GoodEntryInserted); + (await rig.Updater.DeleteRawAsync("big", T0, T0.AddSeconds(2500), default)).ShouldBe(Good); + var live = await RowsAsync(rig, "big", liveOnly: true); + live.Count.ShouldBe(1); + live[0].Num.ShouldBe(-1d); + (await rig.Updater.DeleteRawAsync("big", T0, T0.AddSeconds(2500), default)).ShouldBe(HistorianStatusCodes.BadNoData); + } + + [Fact] + public async Task DeleteRaw_with_a_pre_1970_start_reads_from_the_bound_unclamped() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1), V(T0.AddSeconds(1), 2)], default); + (await rig.Updater.DeleteRawAsync("t", new DateTime(1900, 1, 1, 0, 0, 0, DateTimeKind.Utc), T0.AddSeconds(1), default)) + .ShouldBe(Good); + (await RowsAsync(rig, "t", liveOnly: true)).Select(r => r.Num).ShouldBe([2d]); + // DateTime.MinValue as the start is nudged inside, not bound as -infinity + (await rig.Updater.DeleteRawAsync("t", DateTime.MinValue, DateTime.MaxValue, default)).ShouldBe(Good); + (await RowsAsync(rig, "t", liveOnly: true)).ShouldBeEmpty(); + } + + [Fact] + public async Task Read_after_write_is_visible_across_many_immediate_calls() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + // ~15% of un-waited inserts are not yet visible (spike Q4), so 40 round trips expose a missing wait. + for (var i = 0; i < 40; i++) + { + var t = T0.AddSeconds(i); + (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(t, i)], default)) + .ShouldBe([HistorianStatusCodes.GoodEntryInserted]); + (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [V(t, i + 1000)], default)) + .ShouldBe([HistorianStatusCodes.GoodEntryReplaced], $"iteration {i}: the insert was not visible"); + } + } +}