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

perf(sql): rate-limit per-poll warnings and cache poll plans per reference list (R06:S-6, S-7)

Browse Source 
S-6: a frozen group's BadTimeout warning and a failing group's query-error warning
(with SQL text) logged every poll per group indefinitely; both are now rate-limited
per plan (once a minute), and an uncoercible cell warns once per tag instead of once
per poll. Every poll still publishes the Bad status.
S-7: plans and the slot map were rebuilt every poll although a poll loop passes the
same reference list each cycle. They are cached per list instance (weakly keyed) and
rebuilt only when a reference resolves to a different definition instance, which a
catalog-gate swap produces.

Refs #744
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 16:00:25 -04:00
parent 0a5aa9fc43
commit ab29b76be0
2 changed files with 264 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs
+144 -39
View File
@@ -1,5 +1,7 @@
using System.Collections.Concurrent;
using System.Data.Common;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Globalization;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
@@ -68,6 +70,27 @@ public sealed class SqlPollReader : IDisposable
private long _lastContractWarningTicks;
/// <summary>
/// Per-key last-warned instant for the per-poll warnings (R06:S-6): a frozen group's BadTimeout and a
/// failing group's query error used to log every poll, per group, indefinitely — 20 groups at 1 s was
/// 20 lines a second carrying SQL text. Keyed by the plan's SQL text, so bounded by the plan count.
/// </summary>
private readonly ConcurrentDictionary<string, long> _lastWarnedTicks = new(StringComparer.Ordinal);
/// <summary>Tags whose uncoercible cell has been warned about — once per tag per reader (the refusal rule).</summary>
private readonly ConcurrentDictionary<string, byte> _coerceWarned = new(StringComparer.Ordinal);
/// <summary>
/// The resolved definitions, slot map and plans for each reference list a caller polls (R06:S-7). A
/// poll loop passes the SAME list instance every cycle, so it is planned once instead of every poll.
/// Weakly keyed: an unsubscribed list (and its entry) is collected with it. Invalidation is structural —
/// each entry remembers the definition INSTANCES it was built from, and any reference resolving to a
/// different instance (a catalog-gate swap) rebuilds it.
/// </summary>
private readonly ConditionalWeakTable<IReadOnlyList<string>, PollShape> _shapes = new();
private int _plansBuilt;
/// <summary>Initializes a new instance of the <see cref="SqlPollReader"/> class.</summary>
/// <param name="factory">
/// Creates the provider's connections. Passed explicitly rather than taken from
@@ -182,28 +205,18 @@ public sealed class SqlPollReader : IDisposable
if (results.Length == 0) return new SqlPollResult(results, 0);
var readAt = DateTime.UtcNow;
var shape = GetShape(fullReferences);
// Resolve first. A miss is this tag's own problem (BadNodeIdUnknown) and must not reach the planner,
// whose members are the sole thing the slice-back walks.
var definitions = new List<SqlTagDefinition>(fullReferences.Count);
var slots = new Dictionary<SqlTagDefinition, List<int>>();
for (var i = 0; i < fullReferences.Count; i++)
// A miss is this tag's own problem (BadNodeIdUnknown) and never reaches the planner, whose members
// are the sole thing the slice-back walks.
for (var i = 0; i < shape.Resolved.Length; i++)
{
var definition = fullReferences[i] is { } reference ? _resolve(reference) : null;
if (definition is null)
{
if (shape.Resolved[i] is null)
results[i] = new DataValueSnapshot(null, SqlStatusCodes.BadNodeIdUnknown, null, readAt);
continue;
}
definitions.Add(definition);
if (!slots.TryGetValue(definition, out var positions))
slots[definition] = positions = [];
positions.Add(i);
}
var rowCapExceeded = definitions.Count > 0
? await ReadGroupsAsync(definitions, slots, results, readAt, cancellationToken).ConfigureAwait(false)
var rowCapExceeded = shape.Definitions.Count > 0
? await ReadGroupsAsync(shape, results, readAt, cancellationToken).ConfigureAwait(false)
: 0;
// Fail-safe. Nothing above should leave a hole, but "N in, N out" is a contract the caller indexes
@@ -214,30 +227,93 @@ public sealed class SqlPollReader : IDisposable
return new SqlPollResult(results, rowCapExceeded);
}
/// <summary>How many times a reference list has been (re)planned. Test observation of the S-7 cache.</summary>
internal int PlansBuiltForTests => Volatile.Read(ref _plansBuilt);
/// <summary>
/// Plans the resolved definitions, runs every group concurrently, and slices the results back.
/// Returns how many groups came back over their row cap.
/// Returns the cached shape for <paramref name="fullReferences"/>, rebuilding it when any reference now
/// resolves to a different definition instance than the one it was built from.
/// </summary>
private async Task<int> ReadGroupsAsync(
List<SqlTagDefinition> definitions,
Dictionary<SqlTagDefinition, List<int>> slots,
DataValueSnapshot[] results,
DateTime readAt,
CancellationToken cancellationToken)
private PollShape GetShape(IReadOnlyList<string> fullReferences)
{
IReadOnlyList<SqlQueryPlan> plans;
if (_shapes.TryGetValue(fullReferences, out var cached) && cached.Resolved.Length == fullReferences.Count)
{
var current = true;
for (var i = 0; i < fullReferences.Count && current; i++)
{
var definition = fullReferences[i] is { } reference ? _resolve(reference) : null;
current = ReferenceEquals(definition, cached.Resolved[i]);
}
if (current) return cached;
}
var shape = BuildShape(fullReferences);
_shapes.AddOrUpdate(fullReferences, shape);
return shape;
}
/// <summary>Resolves, slots and plans one reference list.</summary>
private PollShape BuildShape(IReadOnlyList<string> fullReferences)
{
Interlocked.Increment(ref _plansBuilt);
var resolved = new SqlTagDefinition?[fullReferences.Count];
var definitions = new List<SqlTagDefinition>(fullReferences.Count);
var slots = new Dictionary<SqlTagDefinition, List<int>>();
for (var i = 0; i < fullReferences.Count; i++)
{
var definition = fullReferences[i] is { } reference ? _resolve(reference) : null;
resolved[i] = definition;
if (definition is null) continue;
definitions.Add(definition);
if (!slots.TryGetValue(definition, out var positions))
slots[definition] = positions = [];
positions.Add(i);
}
if (definitions.Count == 0) return new PollShape(resolved, definitions, slots, [], PlanningFailed: false);
try
{
plans = SqlGroupPlanner.Plan(definitions, _dialect);
return new PollShape(resolved, definitions, slots, SqlGroupPlanner.Plan(definitions, _dialect), PlanningFailed: false);
}
catch (Exception ex) when (ex is ArgumentException or NotSupportedException)
{
// A definition with a broken invariant — SqlEquipmentTagParser rejects all of these up front, so
// reaching here means a definition was built past the parser. It is an authoring fault, not a
// database fault: Bad-code the planned tags rather than throwing and triggering connection backoff.
// Logged once per (re)plan, which the cache makes once per shape rather than once per poll.
_logger.LogError(ex, "Sql poll planning failed for {Count} tag(s); publishing BadConfigurationError.",
definitions.Count);
foreach (var definition in definitions)
return new PollShape(resolved, definitions, slots, [], PlanningFailed: true);
}
}
/// <summary>One reference list's resolved definitions, slot map and plans — immutable once built.</summary>
private sealed record PollShape(
SqlTagDefinition?[] Resolved,
List<SqlTagDefinition> Definitions,
Dictionary<SqlTagDefinition, List<int>> Slots,
IReadOnlyList<SqlQueryPlan> Plans,
bool PlanningFailed);
/// <summary>
/// Runs every planned group concurrently and slices the results back. Returns how many groups came
/// back over their row cap.
/// </summary>
private async Task<int> ReadGroupsAsync(
PollShape shape,
DataValueSnapshot[] results,
DateTime readAt,
CancellationToken cancellationToken)
{
var slots = shape.Slots;
var plans = shape.Plans;
if (shape.PlanningFailed)
{
foreach (var definition in shape.Definitions)
{
if (!slots.TryGetValue(definition, out var positions)) continue;
var snapshot = new DataValueSnapshot(null, SqlStatusCodes.BadConfigurationError, null, readAt);
@@ -429,11 +505,14 @@ public sealed class SqlPollReader : IDisposable
{
// The table is named for the same reason the contract warnings name it: under a partial outage this
// line repeats every poll, and "which source is frozen" is the only question the operator has.
_logger.LogWarning(
"Sql poll group ({Model}, '{Table}', {Members} tag(s)) exceeded the {Timeout} ms operation " +
"timeout; publishing BadTimeout.",
plan.Model, plan.Members[0].Table, plan.Members.Count,
(int)_operationTimeout.TotalMilliseconds);
if (ShouldWarn("timeout:" + plan.SqlText))
{
_logger.LogWarning(
"Sql poll group ({Model}, '{Table}', {Members} tag(s)) exceeded the {Timeout} ms operation " +
"timeout; publishing BadTimeout. Repeats for this group are suppressed for {Interval} s.",
plan.Model, plan.Members[0].Table, plan.Members.Count,
(int)_operationTimeout.TotalMilliseconds, (int)ContractWarningInterval.TotalSeconds);
}
AbandonedTask.Detach(work);
return GroupResult.TimedOut;
@@ -495,9 +574,13 @@ public sealed class SqlPollReader : IDisposable
{
// The connection opened, so the server is up — this query is what failed (an identifier that
// is not in the catalog, a lock, a permission). That is this group's problem alone.
_logger.LogWarning(
ex, "Sql poll group ({Model}, {Members} tag(s)) failed: {Sql}",
plan.Model, plan.Members.Count, plan.SqlText);
if (ShouldWarn("failed:" + plan.SqlText))
{
_logger.LogWarning(
ex, "Sql poll group ({Model}, {Members} tag(s)) failed: {Sql}. Repeats for this group are " +
"suppressed for {Interval} s.",
plan.Model, plan.Members.Count, plan.SqlText, (int)ContractWarningInterval.TotalSeconds);
}
return GroupResult.Failed;
}
}
@@ -746,9 +829,14 @@ public sealed class SqlPollReader : IDisposable
var target = member.DeclaredType ?? _dialect.MapColumnType(outcome.TypeNames[valueAt]);
if (!TryCoerce(cell, target, out var value))
{
_logger.LogWarning(
"Sql tag '{Tag}' could not coerce a {Actual} cell to its {Declared} type.",
member.Name, cell.GetType().Name, target);
// A per-tag refusal: warned ONCE per tag (per reader), never per poll (R06:S-6).
if (_coerceWarned.TryAdd(member.Name, 0))
{
_logger.LogWarning(
"Sql tag '{Tag}' could not coerce a {Actual} cell to its {Declared} type; it publishes " +
"BadTypeMismatch (warned once per tag).",
member.Name, cell.GetType().Name, target);
}
return new DataValueSnapshot(null, SqlStatusCodes.BadTypeMismatch, sourceTimestamp, readAt);
}
@@ -815,6 +903,23 @@ public sealed class SqlPollReader : IDisposable
Convert.ToDateTime(cell, CultureInfo.InvariantCulture), DateTimeKind.Utc),
};
/// <summary>Rate-limits one per-poll warning by key; true at most once per <see cref="ContractWarningInterval"/>.</summary>
private bool ShouldWarn(string key)
{
var now = Stopwatch.GetTimestamp();
while (true)
{
if (!_lastWarnedTicks.TryGetValue(key, out var last))
{
if (_lastWarnedTicks.TryAdd(key, now)) return true;
continue;
}
if (Stopwatch.GetElapsedTime(last, now) < ContractWarningInterval) return false;
if (_lastWarnedTicks.TryUpdate(key, now, last)) return true;
}
}
/// <summary>Rate-limits the source-contract warning so a misconfigured table cannot flood the log.</summary>
private bool ShouldWarnAboutContract()
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs
+120
View File
@@ -605,6 +605,126 @@ public sealed class SqlPollReaderTests
dialect.Factory, " ", dialect, CommandTimeout, OperationTimeout, 4, false, _ => null));
}
// ---- R06:S-7: plans are cached per subscription reference list ----
/// <summary>
/// A poll loop hands the reader the SAME reference list every cycle; re-planning it (and rebuilding the
/// slot dictionary keyed by a 12-field record) every poll was pure per-poll garbage. Planned once,
/// and the cached plan still reads correct values every time — the positive control.
/// </summary>
[Fact]
public async Task ReadAsync_theSameReferenceList_isPlannedOnce()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture,
KvTag("Speed", SqlitePollFixture.PresentKey), KvTag("Pressure", SqlitePollFixture.SecondPresentKey));
IReadOnlyList<string> refs = ["Speed", "Pressure"];
for (var i = 0; i < 3; i++)
{
var snapshots = await reader.ReadAsync(refs, CancellationToken.None);
snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
snapshots[1].Value.ShouldBe(SqlitePollFixture.SecondPresentValue);
}
reader.PlansBuiltForTests.ShouldBe(1);
}
/// <summary>
/// The cache is invalidated by the definitions themselves: a catalog-gate swap (or any re-resolve)
/// hands back different definition instances, and the next poll re-plans against them.
/// </summary>
[Fact]
public async Task ReadAsync_whenAReferenceResolvesToADifferentDefinition_rePlans()
{
using var fixture = new SqlitePollFixture();
var table = new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal)
{
["Speed"] = KvTag("Speed", SqlitePollFixture.PresentKey),
};
var reader = new SqlPollReader(fixture.Factory, fixture.ConnectionString, new SqliteDialect(),
commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
maxConcurrentGroups: 4, nullIsBad: false, resolve: r => table.GetValueOrDefault(r));
IReadOnlyList<string> refs = ["Speed"];
(await reader.ReadAsync(refs, CancellationToken.None))[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
table["Speed"] = KvTag("Speed", SqlitePollFixture.SecondPresentKey); // re-pointed by a new resolve
(await reader.ReadAsync(refs, CancellationToken.None))[0].Value.ShouldBe(SqlitePollFixture.SecondPresentValue);
reader.PlansBuiltForTests.ShouldBe(2);
}
// ---- R06:S-6: per-poll warnings are rate-limited ----
/// <summary>
/// A dropped table polled every second used to log a warning — with the full SQL text — on every
/// poll, per group, forever. One per minute per plan now; every poll still publishes the Bad status.
/// The first poll's warning is the positive control.
/// </summary>
[Fact]
public async Task ReadAsync_aPersistentlyFailingGroup_warnsOncePerInterval()
{
using var fixture = new SqlitePollFixture();
var logger = new RecordingLogger();
var reader = Reader(fixture, logger,
new SqlTagDefinition("Gone", SqlTagModel.KeyValue, "NoSuchTable",
KeyColumn: SqlitePollFixture.KeyColumn, KeyValue: "k", ValueColumn: SqlitePollFixture.ValueColumn));
for (var i = 0; i < 3; i++)
{
(await reader.ReadAsync(["Gone"], CancellationToken.None))[0]
.StatusCode.ShouldBe(SqlStatusCodes.BadCommunicationError);
}
logger.Entries.Count(e => e.Level == LogLevel.Warning).ShouldBe(1);
}
/// <summary>A cell that will not coerce is a per-tag refusal: one warning per tag, not one per poll.</summary>
[Fact]
public async Task ReadAsync_anUncoercibleCell_warnsOncePerTag()
{
using var fixture = new SqlitePollFixture();
var logger = new RecordingLogger();
var reader = Reader(fixture, logger,
KvTag("Speed", SqlitePollFixture.PresentKey, declaredType: DriverDataType.DateTime));
for (var i = 0; i < 3; i++)
{
(await reader.ReadAsync(["Speed"], CancellationToken.None))[0]
.StatusCode.ShouldBe(SqlStatusCodes.BadTypeMismatch);
}
logger.Entries.Count(e => e.Level == LogLevel.Warning).ShouldBe(1);
}
/// <summary>A frozen group's BadTimeout warning is rate-limited too; both polls still time out.</summary>
[Fact]
public async Task ReadAsync_aRepeatedlyTimedOutGroup_warnsOncePerInterval()
{
using var fixture = new SqlitePollFixture();
var logger = new RecordingLogger();
var reader = new SqlPollReader(fixture.Factory, fixture.ConnectionString, new SqliteDialect(),
commandTimeout: TimeSpan.FromSeconds(30), operationTimeout: TimeSpan.FromMilliseconds(500),
maxConcurrentGroups: 4, nullIsBad: false,
resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey)), logger: logger);
var locker = fixture.OpenNewConnection();
Execute(locker, "BEGIN EXCLUSIVE");
try
{
(await reader.ReadAsync(["Speed"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout);
(await reader.ReadAsync(["Speed"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout);
}
finally
{
Execute(locker, "ROLLBACK");
locker.Dispose();
}
logger.Entries.Count(e => e.Level == LogLevel.Warning && e.Message.Contains("operation timeout"))
.ShouldBe(1);
}
// ---- R06:S-3: the concurrency gate is released ----
/// <summary>An idle reader releases its gate on Dispose, and a disposed reader starts no new group.</summary>