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.