diff --git a/docs/Historian.md b/docs/Historian.md index cdf7a6d2..6bf5dd59 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -839,11 +839,16 @@ data. refusing; doubling the look-back allowance does not skip that final usable budget. Reverse paging retries a crowded segment with a smaller in-window budget derived from the learned ceiling and the observed pre-window rows. -**Missing tags (#653).** Gateway Raw, Processed and AtTime first check the tag catalog. An empty exact-name catalog result or an explicit +**Missing tags (#653).** Gateway Raw, Processed and AtTime first check that the tag exists. An absent tag or an explicit `NotFound` returns no data without a history RPC: Raw is empty; AtTime and Processed return null/`BadNoData` at the requested times or bucket timestamps. Unsupported aggregate validation still runs first. A bounded per-data-source cache -holds at most 4,096 tag names, with **5 seconds** for absence and **60 seconds** for presence, so newly provisioned tags become -visible after the short negative expiry. Catalog faults and cancellation never become missing-tag cache entries. The adapter drains the filtered catalog stream and checks exact names, avoiding older gateways that misparse a missing `GetTagMetadata` response as `Internal`. Empty +holds at most 4,096 tag names, with **5 seconds** for absence and **5 minutes** for presence (was 60 s), so newly provisioned tags become +visible after the short negative expiry; when full it drops expired entries first, else the one closest to expiry (it used to drop an +arbitrary one, which thrashed a sweep over more than 4,096 tags). Lookup faults and cancellation never become missing-tag cache entries. +The check is **one exact unary `GetTagMetadata`** (null = absent). The deployed gateway misparses that reply for a *missing* tag and +answers `Internal`; only that answer (or `Unknown` / the parser's `InvalidDataException`) falls back to the filtered catalog stream +(`BrowseTagNames`, drained, exact-name compare — the filter is a glob, so a name with `*`/`?` streams every match but still answers +exactly). Any other failure propagates and is never turned into absence. Empty AtTime/Processed requests need no check. Events and modified-history capability checks are unaffected. **Stopped-tag reverse discovery (#696).** On a strict Gateway, a truncated remainder probe retains its oldest-row evidence diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs index 8baa06bb..10650775 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayTagPresenceCache.cs @@ -2,10 +2,17 @@ using Grpc.Core; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; -/// Bounded metadata cache. Missing tags expire quickly; failures never become missing tags. +/// +/// Bounded presence cache in front of the gateway's exact tag lookup. A present tag is trusted for +/// , a missing one re-checked after (so a freshly provisioned tag is +/// visible quickly); failures never become missing tags. When full, expired entries are dropped first, else the entry +/// closest to expiry — never an arbitrary one, which thrashed a sweep over more than tags. +/// internal sealed class GatewayTagPresenceCache(IHistorianGatewayClient client, TimeProvider time) { internal const int Capacity = 4096; + internal static readonly TimeSpan PresentTtl = TimeSpan.FromMinutes(5); + internal static readonly TimeSpan MissingTtl = TimeSpan.FromSeconds(5); private readonly object _sync = new(); private readonly Dictionary _entries = new(StringComparer.Ordinal); @@ -25,12 +32,34 @@ internal sealed class GatewayTagPresenceCache(IHistorianGatewayClient client, Ti ct.ThrowIfCancellationRequested(); lock (_sync) { - if (!_entries.ContainsKey(tag) && _entries.Count >= Capacity) _entries.Remove(_entries.Keys.First()); - _entries[tag] = (exists, time.GetUtcNow().AddSeconds(exists ? 60 : 5)); + var now = time.GetUtcNow(); + if (!_entries.ContainsKey(tag) && _entries.Count >= Capacity) MakeRoom(now); + _entries[tag] = (exists, now + (exists ? PresentTtl : MissingTtl)); } return exists; } + /// Drops every expired entry; if none had expired, the one closest to expiry. Called under the lock, when full. + private void MakeRoom(DateTimeOffset now) + { + string? soonest = null; + var soonestExpiry = DateTimeOffset.MaxValue; + List? expired = null; + foreach (var (key, entry) in _entries) + { + if (entry.Expires <= now) (expired ??= []).Add(key); + else if (entry.Expires < soonestExpiry) { soonest = key; soonestExpiry = entry.Expires; } + } + if (expired is not null) foreach (var key in expired) _entries.Remove(key); + else if (soonest is not null) _entries.Remove(soonest); + } + + /// Entries currently held — test/diagnostic. + internal int Count { get { lock (_sync) return _entries.Count; } } + + /// Whether has an entry (fresh or not) — test/diagnostic. + internal bool Holds(string tag) { lock (_sync) return _entries.ContainsKey(tag); } + private static bool IsNotFound(Exception ex) { for (Exception? e = ex; e is not null; e = e.InnerException) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs index d51ef75e..2a716ff6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/HistorianGatewayClientAdapter.cs @@ -82,16 +82,39 @@ public sealed class HistorianGatewayClientAdapter : IHistorianGatewayClient, IDi } /// + // One exact unary lookup (GetTagMetadata: null = absent). The deployed gateway's SDK misparses the reply for a MISSING + // tag and surfaces it as Internal ("GetTagInfo response ended unexpectedly"); only that answer falls back to the tag + // catalog, which proves presence by exact name (BrowseTagNames takes a glob, so a name holding '*' or '?' would stream + // every match — the exact compare keeps it correct). Any other failure propagates: it is never turned into absence. public async Task TagExistsAsync(string tag, CancellationToken ct) { - // older gateway SDKs misparse a missing GetTagInfo response as Internal. A catalog query proves - // absence without guessing from an error. Even a glob-like tag is checked by exact name, never by row count. + try + { + return await _inner.GetTagMetadataAsync(tag, ct).ConfigureAwait(false) is not null; + } + catch (Exception ex) when (!ct.IsCancellationRequested && IsMetadataMisparse(ex)) + { + // Fall through to the catalog. + } + var found = false; await foreach (var name in _inner.BrowseTagNamesAsync(tag, ct).ConfigureAwait(false)) found |= string.Equals(name, tag, StringComparison.OrdinalIgnoreCase); return found; // drain first: a trailing stream fault must propagate and must not become a cache entry } + /// The known metadata misparse: an Internal/Unknown status, or the parser's own InvalidDataException. + private static bool IsMetadataMisparse(Exception ex) + { + for (Exception? e = ex; e is not null; e = e.InnerException) + { + if (e is System.IO.InvalidDataException) return true; + if (e is Grpc.Core.RpcException rpc) + return rpc.StatusCode is Grpc.Core.StatusCode.Internal or Grpc.Core.StatusCode.Unknown; + } + return false; + } + /// public IAsyncEnumerable ReadRawAsync( string tag, DateTime startUtc, DateTime endUtc, int maxValues, CancellationToken ct) => diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs index d47b853c..843c1302 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs @@ -67,13 +67,13 @@ public sealed class GatewayMissingTagTests } [Fact] - public async Task Positive_cache_avoids_checks_per_page_and_expires_after_a_minute() + public async Task Positive_cache_avoids_checks_per_page_and_expires_after_its_ttl() { var fake = new FakeHistorianGatewayClient(); var clock = new Clock(); var ds = Source(fake, clock); await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct); await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct); fake.TagExistsCallCount.ShouldBe(1); - clock.Current = Now.AddSeconds(60); + clock.Current = Now + GatewayTagPresenceCache.PresentTtl; await ds.ReadRawAsync("T", Now.AddMinutes(-1), Now, 10, Ct); fake.TagExistsCallCount.ShouldBe(2); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs index 23fb00f2..045ca1f9 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs @@ -21,17 +21,42 @@ public sealed class GatewayTagCatalogWireTests [InlineData("Pump1", true)] [InlineData("Missing", false)] [InlineData("Pump*", false)] - public async Task Presence_is_proved_by_exact_names_without_using_the_broken_metadata_parser(string tag, bool expected) + public async Task Presence_is_one_exact_metadata_lookup_with_no_catalog_browse(string tag, bool expected) { - var service = new TagsService(); + var service = new TagsService { MetadataWorks = true }; await using var server = await StartAsync(service); using var adapter = Create(server.Port); (await adapter.TagExistsAsync(tag, Ct)).ShouldBe(expected); + service.MetadataCalls.ShouldBe(1); + service.LastMetadataTag.ShouldBe(tag); + service.BrowseCalls.ShouldBe(0, "an exact lookup needs no glob-filtered catalog stream"); + } + + [Theory] + [InlineData("Pump1", true)] + [InlineData("Missing", false)] + [InlineData("Pump*", false)] + public async Task A_gateway_that_misparses_metadata_falls_back_to_the_exact_named_catalog(string tag, bool expected) + { + var service = new TagsService(); // the deployed gateway: a missing GetTagInfo reply surfaces as Internal + await using var server = await StartAsync(service); + using var adapter = Create(server.Port); + (await adapter.TagExistsAsync(tag, Ct)).ShouldBe(expected); + service.MetadataCalls.ShouldBe(1); service.Filter.ShouldBe(tag); - service.MetadataCalls.ShouldBe(0); service.Completed.ShouldBe(1); } + [Fact] + public async Task A_metadata_transport_failure_propagates_and_is_never_turned_into_absence() + { + var service = new TagsService { MetadataStatus = StatusCode.PermissionDenied }; + await using var server = await StartAsync(service); + using var adapter = Create(server.Port); + await Should.ThrowAsync(() => adapter.TagExistsAsync("Pump1", Ct)); + service.BrowseCalls.ShouldBe(0, "only the known metadata misparse falls back to the catalog"); + } + [Fact] public async Task A_catalog_stream_fault_after_a_matching_name_is_not_cached_as_presence() { @@ -53,13 +78,22 @@ public sealed class GatewayTagCatalogWireTests private sealed class TagsService : HistorianTags.HistorianTagsBase { public int MetadataCalls, BrowseCalls, Completed; - public string? Filter; - public bool FailAfterRow; + public string? Filter, LastMetadataTag; + public bool FailAfterRow, MetadataWorks; + public StatusCode MetadataStatus = StatusCode.Internal; public string[] Names = ["Pump1", "Pump2"]; public override Task GetTagMetadata(GetTagMetadataRequest request, ServerCallContext context) { MetadataCalls++; - throw new RpcException(new Status(StatusCode.Internal, "GetTagInfo response ended unexpectedly.")); + LastMetadataTag = request.Tag; + if (MetadataWorks) + { + // A fixed gateway: exact name, metadata only when the tag exists (proto3 optional). + var reply = new GetTagMetadataReply(); + if (Names.Contains(request.Tag, StringComparer.Ordinal)) reply.Metadata = new HistorianTagMetadata { Name = request.Tag }; + return Task.FromResult(reply); + } + throw new RpcException(new Status(MetadataStatus, "GetTagInfo response ended unexpectedly.")); } public override async Task BrowseTagNames(BrowseTagNamesRequest request, IServerStreamWriter stream, ServerCallContext context) { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs index 90feb181..3e9b5ad2 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs @@ -42,7 +42,7 @@ public sealed class GatewayTagPresenceCacheConcurrencyTests(ITestOutputHelper ou [Theory] [InlineData(false, 5)] - [InlineData(true, 60)] + [InlineData(true, 300)] // GatewayTagPresenceCache.PresentTtl (5 min) public async Task Overlapping_expired_lookups_refresh_presence_at_the_TTL_boundary(bool initial, int ttlSeconds) { // Catches an inclusive expiry comparison or failure to replace an expired positive/negative entry. @@ -161,7 +161,7 @@ public sealed class GatewayTagPresenceCacheConcurrencyTests(ITestOutputHelper ou [Theory] [InlineData(false, 5)] - [InlineData(true, 60)] + [InlineData(true, 300)] // GatewayTagPresenceCache.PresentTtl (5 min) public async Task An_expired_entry_is_rechecked_after_a_faulting_refresh(bool initial, int ttlSeconds) { // Catches faults renewing an expired negative or replacing an expired positive with a negative. @@ -184,7 +184,7 @@ public sealed class GatewayTagPresenceCacheConcurrencyTests(ITestOutputHelper ou [Theory] [InlineData(false, 5)] - [InlineData(true, 60)] + [InlineData(true, 300)] // GatewayTagPresenceCache.PresentTtl (5 min) public async Task An_expired_entry_is_rechecked_after_a_cancelled_refresh(bool initial, int ttlSeconds) { // Catches cancellation renewing expired entries when the catalog returns an absence result anyway. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheEvictionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheEvictionTests.cs new file mode 100644 index 00000000..43f76646 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheEvictionTests.cs @@ -0,0 +1,77 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// The presence cache keeps a present tag for minutes (one exact lookup per tag per TTL, not one per HistoryRead), and +/// a full cache evicts what is expired or closest to expiry — never an arbitrary entry. +/// +public sealed class GatewayTagPresenceCacheEvictionTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private sealed class Clock : TimeProvider + { + public DateTimeOffset Now = new(2026, 10, 22, 0, 0, 0, TimeSpan.Zero); + public override DateTimeOffset GetUtcNow() => Now; + } + + [Fact] + public async Task A_present_tag_is_served_from_the_cache_until_its_ttl_then_re_checked() + { + var clock = new Clock(); + var fake = new FakeHistorianGatewayClient { TagExistsResult = true }; + var cache = new GatewayTagPresenceCache(fake, clock); + + (await cache.ExistsAsync("A", Ct)).ShouldBeTrue(); + clock.Now += GatewayTagPresenceCache.PresentTtl - TimeSpan.FromSeconds(1); + (await cache.ExistsAsync("A", Ct)).ShouldBeTrue(); + fake.TagExistsCallCount.ShouldBe(1, "a fresh present entry needs no lookup"); + + clock.Now += TimeSpan.FromSeconds(2); + (await cache.ExistsAsync("A", Ct)).ShouldBeTrue(); + fake.TagExistsCallCount.ShouldBe(2, "an expired entry is re-checked"); + } + + [Fact] + public async Task A_full_cache_evicts_the_entry_closest_to_expiry_not_the_oldest_present_one() + { + var clock = new Clock(); + var fake = new FakeHistorianGatewayClient { TagExistsResult = true }; + var cache = new GatewayTagPresenceCache(fake, clock); + + // The first-inserted entry is a PRESENT tag the old Keys.First() eviction would have dropped. + await cache.ExistsAsync("first-present", Ct); + clock.Now += TimeSpan.FromSeconds(1); + fake.TagExistsResult = false; + await cache.ExistsAsync("missing", Ct); // expires in 5 s — the closest to expiry + fake.TagExistsResult = true; + for (var i = 0; cache.Count < GatewayTagPresenceCache.Capacity; i++) await cache.ExistsAsync($"t{i}", Ct); + + await cache.ExistsAsync("newcomer", Ct); + + cache.Count.ShouldBe(GatewayTagPresenceCache.Capacity); + cache.Holds("newcomer").ShouldBeTrue(); + cache.Holds("missing").ShouldBeFalse("the entry closest to expiry goes first"); + cache.Holds("first-present").ShouldBeTrue("an arbitrary (first-inserted) entry is never the victim"); + } + + [Fact] + public async Task A_full_cache_drops_every_expired_entry_before_evicting_a_live_one() + { + var clock = new Clock(); + var fake = new FakeHistorianGatewayClient { TagExistsResult = false }; + var cache = new GatewayTagPresenceCache(fake, clock); + for (var i = 0; i < 10; i++) await cache.ExistsAsync($"gone{i}", Ct); // missing ⇒ 5 s + fake.TagExistsResult = true; + for (var i = 0; cache.Count < GatewayTagPresenceCache.Capacity; i++) await cache.ExistsAsync($"t{i}", Ct); + + clock.Now += TimeSpan.FromSeconds(10); // the ten missing entries have expired; the present ones have not + await cache.ExistsAsync("newcomer", Ct); + + cache.Count.ShouldBe(GatewayTagPresenceCache.Capacity - 10 + 1); + for (var i = 0; i < 10; i++) cache.Holds($"gone{i}").ShouldBeFalse(); + cache.Holds("t0").ShouldBeTrue(); + } +}