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 index c818e8b0..27e4f416 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs @@ -15,6 +15,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// "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). +/// Concurrency: calls are serialized per tag inside this instance (64 striped async locks held from +/// the existence read through the visibility wait), so concurrent Insert/Replace/Delete on one tag cannot +/// both act on a stale read. This does NOT cover the live writer racing on the same key, nor a second node; +/// that residual is accepted. /// 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 @@ -27,7 +31,13 @@ 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; + /// Request times further apart than this are read with separate range queries (I-2). + internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1); + private const long ResolutionTicks = 10; + private const int LockStripes = 64; + private readonly SemaphoreSlim[] _locks = Enumerable.Range(0, LockStripes).Select(_ => new SemaphoreSlim(1, 1)).ToArray(); + private long _liveRowsRead; private readonly QuestDbConnection _conn; private readonly QuestDbTagCatalog _tags; private readonly QuestDbWalVisibility _wal; @@ -47,6 +57,41 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater _log = log ?? NullLogger.Instance; } + /// Rows returned by existence reads so far (test/diagnostic counter). + internal long LiveRowsRead => Interlocked.Read(ref _liveRowsRead); + + /// Splits times into clusters wherever consecutive sorted times are further apart than . + /// Request times (any order). + /// Largest gap kept inside one cluster. + /// Inclusive [Lo, Hi] ranges, ascending. + internal static List<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList times, TimeSpan gap) + { + var result = new List<(DateTime, DateTime)>(); + if (times.Count == 0) return result; + var sorted = times.OrderBy(t => t).ToList(); + var lo = sorted[0]; + var prev = lo; + foreach (var t in sorted.Skip(1)) + { + if (t - prev > gap) { result.Add((lo, prev)); lo = t; } + prev = t; + } + result.Add((lo, prev)); + return result; + } + + private async Task LockAsync(string tag, CancellationToken ct) + { + var sem = _locks[(int)((uint)StringComparer.Ordinal.GetHashCode(tag) % LockStripes)]; + await sem.WaitAsync(ct).ConfigureAwait(false); + return new Releaser(sem); + } + + private sealed class Releaser(SemaphoreSlim sem) : IDisposable + { + public void Dispose() => sem.Release(); + } + /// public async Task> UpdateDataAsync(string tag, HistoryUpdateMode mode, IReadOnlyList values, CancellationToken ct) @@ -73,6 +118,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater } else { + using var _ = await LockAsync(tag, ct).ConfigureAwait(false); var known = mode == HistoryUpdateMode.Replace ? await _tags.ExistsAsync(tag, ct).ConfigureAwait(false) : await EnsureAsync(tag, plan.Winners[0].Row.Value.Kind, ct).ConfigureAwait(false); @@ -93,6 +139,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater if (!await _tags.ExistsAsync(tag, ct).ConfigureAwait(false)) return HistoryUpdatePlanner.DeleteRawResult(tagKnown: false, 0); + using var _ = await LockAsync(tag, ct).ConfigureAwait(false); var cursor = QuestDbConnection.Round(startUtc); var end = QuestDbConnection.Round(endUtc); long total = 0; @@ -142,7 +189,8 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater 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); + using var _ = await LockAsync(tag, ct).ConfigureAwait(false); + var live = await ReadLiveAsync(tag, candidates.Select(t => new DateTime(t, DateTimeKind.Utc)).ToList(), 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); @@ -161,8 +209,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater 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 live = await ReadLiveAsync(tag, rows.Select(r => QuestDbConnection.Round(r.TimeUtc)).ToList(), ct).ConfigureAwait(false); var outcome = new Dictionary(); var toWrite = new List(); foreach (var r in rows) @@ -179,18 +226,25 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater 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) + /// Ticks of the live (present, not deleted) rows near the given times: one range query per cluster (I-2). + private async Task> ReadLiveAsync(string tag, IReadOnlyList times, 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); + foreach (var (lo, hi) in Cluster(times, ExistenceClusterGap)) + { + 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, lo); + QuestDbConnection.BindTimestamp(cmd, hi); + await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await rd.ReadAsync(ct).ConfigureAwait(false)) + { + live.Add(QuestDbConnection.ReadUtc(rd, 0).Ticks); + Interlocked.Increment(ref _liveRowsRead); + } + } return live; } 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 index c77f3d64..0f721065 100644 --- 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 @@ -232,4 +232,47 @@ public sealed class UpdateTests(QuestDbFixture fx) .ShouldBe([HistorianStatusCodes.GoodEntryReplaced], $"iteration {i}: the insert was not visible"); } } + + [Fact] + public async Task Concurrent_inserts_of_the_same_new_key_yield_exactly_one_GoodEntryInserted() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + for (var i = 0; i < 20; i++) // loop so the race shows reliably without the lock + { + var t = T0.AddSeconds(i); + var codes = await Task.WhenAll( + Task.Run(() => rig.Updater.UpdateDataAsync("race", HistoryUpdateMode.Insert, [V(t, 1)], default)), + Task.Run(() => rig.Updater.UpdateDataAsync("race", HistoryUpdateMode.Insert, [V(t, 2)], default))); + codes.Select(c => c[0]).OrderBy(c => c).ToArray().ShouldBe( + new[] { HistorianStatusCodes.GoodEntryInserted, HistorianStatusCodes.BadEntryExists }, $"iteration {i}"); + } + } + + [Fact] + public void Cluster_splits_where_the_gap_exceeds_the_limit() + { + var gap = QuestDbHistorianDataUpdater.ExistenceClusterGap; + var c = QuestDbHistorianDataUpdater.Cluster( + [T0.AddHours(2), T0, T0.AddSeconds(30), T0.AddHours(1)], gap); + c.ShouldBe([(T0, T0.AddSeconds(30)), (T0.AddHours(1), T0.AddHours(1)), (T0.AddHours(2), T0.AddHours(2))]); + QuestDbHistorianDataUpdater.Cluster([], gap).ShouldBeEmpty(); + } + + [Fact] + public async Task Existence_reads_are_bounded_when_request_times_are_far_apart() + { + fx.SkipIfUnavailable(); + var rig = MakeRig(fx); + // 1 Hz seed across 2 h = 7,200 live rows; the request touches 3 instants 1 h apart. + var seed = Enumerable.Range(0, 7200).Select(i => V(T0.AddSeconds(i), i)).ToList(); + await rig.Updater.UpdateDataAsync("sparse", HistoryUpdateMode.Insert, seed, default); + var before = rig.Updater.LiveRowsRead; + var codes = await rig.Updater.UpdateDataAsync("sparse", HistoryUpdateMode.Replace, + [V(T0, 1), V(T0.AddHours(1), 2), V(T0.AddHours(2).AddSeconds(-1), 3), V(T0.AddHours(1).AddMilliseconds(500), 4)], default); + codes.ShouldBe([HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.GoodEntryReplaced, + HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.BadNoEntryExists]); + // 3 clusters of a single instant each (T0+1h and +1h.5s are one cluster of 2 instants, 1 live row) + (rig.Updater.LiveRowsRead - before).ShouldBeLessThanOrEqualTo(3); + } }