dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(historian): per-tag update lock + clustered existence reads (T6 review)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-11 02:12:09 -04:00
parent ee174c0179
commit 5c0671e5b5
2 changed files with 109 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs
+66 -12
View File
@@ -15,6 +15,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// "Exists" means present and not deleted, so Insert over a tombstone succeeds (<c>GoodEntryInserted</c>) 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).</para>
/// <para><b>Concurrency:</b> 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.</para>
/// <para><b>Pre-1970 (RC1):</b> QuestDB cannot store such a designated timestamp. Insert/Replace/Update of such an
/// item reports <c>BadOutOfRange</c> and sends nothing (a null/<c>MinValue</c> timestamp is still
/// <c>BadInvalidTimestamp</c>); DeleteAtTime reports <c>BadNoEntryExists</c>; a DeleteRaw window may start before
@@ -27,7 +31,13 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
/// <summary>Rows per INSERT and per DeleteRaw page (T0 Q2: 9,000 parameters, far under the 32,767 limit).</summary>
internal const int BatchSize = 1000;
/// <summary>Request times further apart than this are read with separate range queries (I-2).</summary>
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;
}
/// <summary>Rows returned by existence reads so far (test/diagnostic counter).</summary>
internal long LiveRowsRead => Interlocked.Read(ref _liveRowsRead);
/// <summary>Splits times into clusters wherever consecutive sorted times are further apart than <paramref name="gap"/>.</summary>
/// <param name="times">Request times (any order).</param>
/// <param name="gap">Largest gap kept inside one cluster.</param>
/// <returns>Inclusive [Lo, Hi] ranges, ascending.</returns>
internal static List<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList<DateTime> 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<IDisposable> 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();
}
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> UpdateDataAsync(string tag, HistoryUpdateMode mode,
IReadOnlyList<HistorizationValue> 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<HistorianSampleRow> 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<long, UpsertOutcome>();
var toWrite = new List<HistorianSampleRow>();
foreach (var r in rows)
@@ -179,18 +226,25 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
return outcome;
}
/// <summary>Ticks of the live (present, not deleted) rows with <c>ts</c> in <c>[minTicks, maxTicks]</c>: one range query.</summary>
private async Task<HashSet<long>> ReadLiveAsync(string tag, long minTicks, long maxTicks, CancellationToken ct)
/// <summary>Ticks of the live (present, not deleted) rows near the given times: one range query per cluster (I-2).</summary>
private async Task<HashSet<long>> ReadLiveAsync(string tag, IReadOnlyList<DateTime> times, CancellationToken ct)
{
var live = new HashSet<long>();
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;
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs
+43
View File
@@ -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);
}
}