From ab29b76be0feddc39844e8146230920dc670e543 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 16:00:25 -0400 Subject: [PATCH] perf(sql): rate-limit per-poll warnings and cache poll plans per reference list (R06:S-6, S-7) 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 --- .../SqlPollReader.cs | 183 ++++++++++++++---- .../SqlPollReaderTests.cs | 120 ++++++++++++ 2 files changed, 264 insertions(+), 39 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs index 25e70a39..fdb22ff0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs @@ -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; + /// + /// 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. + /// + private readonly ConcurrentDictionary _lastWarnedTicks = new(StringComparer.Ordinal); + + /// Tags whose uncoercible cell has been warned about — once per tag per reader (the refusal rule). + private readonly ConcurrentDictionary _coerceWarned = new(StringComparer.Ordinal); + + /// + /// 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. + /// + private readonly ConditionalWeakTable, PollShape> _shapes = new(); + + private int _plansBuilt; + /// Initializes a new instance of the class. /// /// 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(fullReferences.Count); - var slots = new Dictionary>(); - 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); } + /// How many times a reference list has been (re)planned. Test observation of the S-7 cache. + internal int PlansBuiltForTests => Volatile.Read(ref _plansBuilt); + /// - /// 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 , rebuilding it when any reference now + /// resolves to a different definition instance than the one it was built from. /// - private async Task ReadGroupsAsync( - List definitions, - Dictionary> slots, - DataValueSnapshot[] results, - DateTime readAt, - CancellationToken cancellationToken) + private PollShape GetShape(IReadOnlyList fullReferences) { - IReadOnlyList 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; + } + + /// Resolves, slots and plans one reference list. + private PollShape BuildShape(IReadOnlyList fullReferences) + { + Interlocked.Increment(ref _plansBuilt); + + var resolved = new SqlTagDefinition?[fullReferences.Count]; + var definitions = new List(fullReferences.Count); + var slots = new Dictionary>(); + 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); + } + } + + /// One reference list's resolved definitions, slot map and plans — immutable once built. + private sealed record PollShape( + SqlTagDefinition?[] Resolved, + List Definitions, + Dictionary> Slots, + IReadOnlyList Plans, + bool PlanningFailed); + + /// + /// Runs every planned group concurrently and slices the results back. Returns how many groups came + /// back over their row cap. + /// + private async Task 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), }; + /// Rate-limits one per-poll warning by key; true at most once per . + 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; + } + } + /// Rate-limits the source-contract warning so a misconfigured table cannot flood the log. private bool ShouldWarnAboutContract() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs index 4181449c..d26bd0b6 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs @@ -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 ---- + + /// + /// 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. + /// + [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 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); + } + + /// + /// 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. + /// + [Fact] + public async Task ReadAsync_whenAReferenceResolvesToADifferentDefinition_rePlans() + { + using var fixture = new SqlitePollFixture(); + var table = new Dictionary(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 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 ---- + + /// + /// 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. + /// + [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); + } + + /// A cell that will not coerce is a per-tag refusal: one warning per tag, not one per poll. + [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); + } + + /// A frozen group's BadTimeout warning is rate-limited too; both polls still time out. + [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 ---- /// An idle reader releases its gate on Dispose, and a disposed reader starts no new group.