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

perf(historian): exact tag lookup and a non-thrashing presence cache (lmxopcua#746)

Browse Source 
Every Gateway HistoryRead was preceded by a BrowseTagNames catalog stream
keyed by a glob filter (a name with '*' or '?' streamed every match), and the
4096-entry presence cache evicted an arbitrary entry (Keys.First()) when full,
so a sweep over more tags thrashed it.

The presence check is now one exact unary GetTagMetadata (null = absent).
Only the deployed gateway's known misparse of a missing tag (Internal /
Unknown / InvalidDataException) falls back to the exact-name catalog check;
any other failure propagates and never becomes absence. The positive TTL is 5
minutes (was 60 s); a full cache drops expired entries first, else the one
closest to expiry. Wire tests cover the working, the misparsing and the
refusing gateway.

Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22
in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-10-22 16:36:24 -04:00
parent 8dc594d8e4
commit bc7e58ed64
7 changed files with 187 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayMissingTagTests.cs
+2 -2
View File
@@ -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);
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagCatalogWireTests.cs
+40 -6
View File
@@ -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<Exception>(() => 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<GetTagMetadataReply> 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<TagName> stream, ServerCallContext context)
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs
+3 -3
View File
@@ -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.
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheEvictionTests.cs
+77
View File
@@ -0,0 +1,77 @@
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
/// <summary>
/// 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.
/// </summary>
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();
}
}