diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index e5c19a87..90ebabd6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -5,6 +5,7 @@ using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; @@ -73,6 +74,23 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// private readonly int _maxRawValuesPerRead; + /// The default of eventDrainBudget: ServerHistorian:Gateway:EventDrainBudget's default. + public const int DefaultEventDrainBudget = GatewayHistorianSettings.DefaultEventDrainBudget; + + /// + /// Stream rows an events read drains past its cap before it abandons the gateway stream; see + /// . + /// + private readonly int _eventDrainBudget; + + private long _eventStreamsAbandoned; + + /// The bound events drain budget — exposed so the factory wiring is testable. + internal int EventDrainBudget => _eventDrainBudget; + + /// Events reads that abandoned their gateway stream past the drain budget (lifetime) — test/diagnostic. + internal long EventStreamsAbandoned => Interlocked.Read(ref _eventStreamsAbandoned); + /// A strict gateway's row ceiling, learned from its first refusal. private readonly GatewayRowCeiling _ceiling; private readonly GatewayTagPresenceCache _tags; @@ -109,10 +127,16 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// for cap + 1 rows and is refused when the window holds more, never truncated. <= 0 = unbounded (the pre-#736 /// behaviour: an uncapped read is sent as ). /// + /// + /// Stream rows a capped events read consumes (uncollected) past its cap before it abandons the gateway stream + /// (ServerHistorian:Gateway:EventDrainBudget). Negative counts as 0 = abandon at the cap. + /// public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger logger, int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null, - bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows, int maxRawValuesPerRead = 0) + bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows, int maxRawValuesPerRead = 0, + int eventDrainBudget = DefaultEventDrainBudget) { + _eventDrainBudget = Math.Max(0, eventDrainBudget); _maxPreWindowRows = Math.Max(1, maxPreWindowRows); _maxRawValuesPerRead = Math.Max(0, maxRawValuesPerRead); _atTimeFromRaw = atTimeFromRaw; @@ -556,6 +580,10 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis RecordOutcome(success: true, error: null); return new HistoryReadResult(aligned, ContinuationPoint: null); } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; // caller-requested, not a backend fault + } catch (Exception ex) { RecordReadFailure(ex); @@ -609,6 +637,13 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis : ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken); /// + // The gateway's ReadEvents has no max-events field (the 0.3.0 wire contract), so the cap is applied here. The stream + // is DRAINED past the cap — counted, never collected — rather than broken out of: abandoning a gateway stream cancels + // the RPC, which the gateway classifies as a connection fault and answers by evicting its pooled historian session + // (the next read of any kind reconnects; concurrent reads on that session fail). The drain is bounded by the + // EventDrainBudget (ServerHistorian:Gateway:EventDrainBudget): past cap + budget events the stream is abandoned after + // all (a window far larger than the page — streaming all of it for every page would cost O(window²/N)), logged at + // Debug and counted in EventStreamsAbandoned. 0 = abandon at the cap (the previous behaviour). public async Task ReadEventsAsync( string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) @@ -619,11 +654,34 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis // Bounded initial capacity: an uncapped read arrives as int.MaxValue ("Array dimensions exceeded"). var collected = new List(hasCap ? Math.Min(maxEvents, 1024) : 0); var truncated = false; + long pastCap = 0; // stream rows consumed after the cap filled (matching or not) + var abandoned = false; await foreach (var wireEvent in _client .ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken) .ConfigureAwait(false)) { + if (hasCap && collected.Count == maxEvents) + { + // The page is full: drain the rest of the stream without collecting it, within the budget. + // One more MATCHING event is the "more events exist" signal; a foreign source's row is not, and + // until the signal is seen no row counts toward the budget — abandoning before it would answer + // "complete" for a page that may not be (and a spurious continuation point on a stretch of foreign + // rows would make the server crawl it one resolution step per page). + if (!truncated) + { + if (sourceName is not null && !string.Equals(wireEvent.SourceName, sourceName, StringComparison.Ordinal)) + continue; + truncated = true; + } + if (++pastCap > _eventDrainBudget) + { + abandoned = true; + break; + } + continue; + } + var mapped = EventMapper.ToHistoricalEvent(wireEvent); // Defensive client-side source filter: the gateway's SQL ReadEvents source filter @@ -633,17 +691,17 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis continue; } - // One more matching event arriving once the cap is full means the result is - // truncated — stop draining and flag it. - if (hasCap && collected.Count == maxEvents) - { - truncated = true; - break; - } - collected.Add(mapped); } + if (abandoned) + { + Interlocked.Increment(ref _eventStreamsAbandoned); + _logger.LogDebug( + "HistorianGateway events read abandoned its stream {Budget} rows past the page cap (EventDrainBudget); " + + "the gateway may evict its historian session.", _eventDrainBudget); + } + RecordOutcome(success: true, error: null); // A non-null, opaque token signals truncation to the caller. // The gateway has no resumable cursor, so the token's contents carry no paging state — @@ -651,6 +709,10 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis // from being shared/mutated. return new HistoricalEventsResult(collected, truncated ? new byte[] { 0x01 } : null); } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; // the caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault + } catch (Exception ex) { RecordReadFailure(ex); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs index a2a9b0d2..ea2ff2ef 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs @@ -46,7 +46,9 @@ public static class GatewayHistorian // A crowding re-read never makes room for more pre-window rows than the tie-cluster limit. maxPreWindowRows: options.MaxTieClusterOverfetch, // The server raw cap (R08:1.1): the page limit the server reads at, and the bound on every raw read. - maxRawValuesPerRead: options.EffectiveMaxRawValuesPerRead); + maxRawValuesPerRead: options.EffectiveMaxRawValuesPerRead, + // How far a capped events read drains its stream past the cap before abandoning it. + eventDrainBudget: options.Gateway.EventDrainBudget); } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs index ba2fe9c7..81d5bdf1 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs @@ -106,6 +106,12 @@ public sealed class ServerHistorianOptions /// public bool GatewayAtTimeFromRaw { get; init; } = true; + /// + /// Gateway backend only: the ServerHistorian:Gateway sub-section — knobs that exist because of the + /// HistorianGateway's wire contract (see ). Ignored by the other backends. + /// + public GatewayHistorianSettings Gateway { get; init; } = new(); + /// /// Opt-in: HistoryReadEvents never reads events older than now − EventReadHorizon — the earlier /// bound of every fresh events window is raised to it (both directions, every backend; a window wholly older is @@ -252,6 +258,28 @@ public sealed class ServerHistorianOptions warnings.Add($"ServerHistorian:EventReadHorizon is {EventReadHorizon} — negative is treated as no horizon (00:00:00); HistoryReadEvents reads every window in full."); if (HistoryReadDeadline <= TimeSpan.Zero) warnings.Add($"ServerHistorian:HistoryReadDeadline is {HistoryReadDeadline} — must be > 0; HistoryRead requests run unbounded (no per-request deadline)."); + if (IsGatewayBackend && Gateway.EventDrainBudget < 0) + warnings.Add($"ServerHistorian:Gateway:EventDrainBudget is {Gateway.EventDrainBudget} — negative is treated as 0: every capped events read abandons its gateway stream at the cap, which makes the gateway evict its historian session."); return warnings; } } + +/// +/// The ServerHistorian:Gateway sub-section: Gateway-backend knobs that exist because of the HistorianGateway's +/// wire contract. Every default is the safe posture. +/// +public sealed class GatewayHistorianSettings +{ + /// The default of . + public const int DefaultEventDrainBudget = 10_000; + + /// + /// How many stream rows a capped HistoryReadEvents read consumes (and discards) past its page cap before it + /// abandons the gateway stream. The gateway's ReadEvents has no max-events field, so the cap is applied + /// client-side; an abandoned stream cancels the RPC and the gateway evicts its pooled historian session (the + /// next read of any kind reconnects, and concurrent reads on that session fail). Draining is cheap for the + /// usual small overshoot; past the budget the stream is abandoned anyway, so a page over a far larger window + /// does not stream the whole window. Default 10 000; 0 = abandon at the cap (the previous behaviour). + /// + public int EventDrainBudget { get; init; } = DefaultEventDrainBudget; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs index 52c526ef..744da36d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs @@ -71,6 +71,13 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient /// When set, ReadEvents serves EventsSource(start, end) instead of . public Func>? EventsSource; + /// Events actually pulled by the consumer across every ReadEvents stream. + public int EventsYieldedCount; + /// ReadEvents streams the consumer enumerated to the END (never broke out of — R08:1.2). + public int EventStreamsCompleted; + /// ReadEvents streams the consumer disposed before their end (an abandoned gateway stream). + public int EventStreamsAbandoned; + /// Thrown (deferred to first enumeration) by every read method when set. public Exception? ThrowOnRead; @@ -204,7 +211,7 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient LastReadEventsEndUtc = endUtc; LastReadEventsMaxEvents = maxEvents; ReadEventsCallCount++; - return ToAsyncStream(EventsSource is null ? Events : EventsSource(startUtc, endUtc).ToList(), ThrowOnRead, ct); + return CountingEventStream(ToAsyncStream(EventsSource is null ? Events : EventsSource(startUtc, endUtc).ToList(), ThrowOnRead, ct)); } public Task WriteLiveValuesAsync( @@ -272,6 +279,26 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient RawStreamsCompleted++; } + private async IAsyncEnumerable CountingEventStream(IAsyncEnumerable inner) + { + var completed = false; + try + { + await foreach (var item in inner.ConfigureAwait(false)) + { + Interlocked.Increment(ref EventsYieldedCount); + yield return item; + } + completed = true; + Interlocked.Increment(ref EventStreamsCompleted); + } + finally + { + // A consumer that breaks out (or a fault) disposes the stream early: on the real gateway that cancels the RPC. + if (!completed) Interlocked.Increment(ref EventStreamsAbandoned); + } + } + private static async IAsyncEnumerable ToAsyncStream( IReadOnlyList items, Exception? error, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayEventDrainTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayEventDrainTests.cs new file mode 100644 index 00000000..b34e760a --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayEventDrainTests.cs @@ -0,0 +1,280 @@ +using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// The Gateway events read drains its stream past the cap (within a budget) instead of abandoning it — an abandoned +/// gateway stream cancels the RPC and the gateway evicts its pooled historian session — and a caller cancellation on +/// the events / native-AtTime arms is not a backend failure. +/// +public sealed class GatewayEventDrainTests +{ + private static readonly DateTime T0 = new(2026, 10, 22, 8, 0, 0, DateTimeKind.Utc); + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + [Fact] + public async Task A_capped_read_drains_the_stream_to_its_end_and_still_flags_truncation() + { + var fake = new FakeHistorianGatewayClient { Events = Series(50) }; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + + var r = await ds.ReadEventsAsync(null, T0, T0.AddHours(1), maxEvents: 10, Ct); + + r.Events.Select(e => e.EventId).ShouldBe(Enumerable.Range(0, 10).Select(i => $"E{i:000}")); + r.ContinuationPoint.ShouldNotBeNull(); + fake.EventStreamsCompleted.ShouldBe(1, "the stream must be consumed to its end, never broken out of"); + fake.EventStreamsAbandoned.ShouldBe(0); + fake.EventsYieldedCount.ShouldBe(50); + ds.EventStreamsAbandoned.ShouldBe(0); + } + + [Fact] + public async Task Past_the_drain_budget_the_stream_is_abandoned_but_the_page_is_unchanged() + { + var fake = new FakeHistorianGatewayClient { Events = Series(100) }; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, eventDrainBudget: 5); + + var r = await ds.ReadEventsAsync(null, T0, T0.AddHours(1), maxEvents: 10, Ct); + + r.Events.Count.ShouldBe(10); + r.ContinuationPoint.ShouldNotBeNull(); + fake.EventStreamsAbandoned.ShouldBe(1); + fake.EventsYieldedCount.ShouldBe(10 + 5 + 1, "cap + budget, then the row that overran it"); + ds.EventStreamsAbandoned.ShouldBe(1); + } + + [Fact] + public async Task Budget_zero_is_the_previous_abandon_at_the_cap() + { + var fake = new FakeHistorianGatewayClient { Events = Series(20) }; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, eventDrainBudget: 0); + + var r = await ds.ReadEventsAsync(null, T0, T0.AddHours(1), maxEvents: 3, Ct); + + r.Events.Count.ShouldBe(3); + r.ContinuationPoint.ShouldNotBeNull(); + fake.EventsYieldedCount.ShouldBe(4); + fake.EventStreamsAbandoned.ShouldBe(1); + } + + [Fact] + public async Task Foreign_source_rows_past_the_cap_are_drained_but_do_not_signal_truncation() + { + // The page is exactly the matching events; everything after it belongs to another source (an unfiltered gateway). + var events = Series(3, "Pump1").Concat(Series(5, "Pump2", offsetSeconds: 10)).ToArray(); + var fake = new FakeHistorianGatewayClient { Events = events }; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + + var r = await ds.ReadEventsAsync("Pump1", T0, T0.AddHours(1), maxEvents: 3, Ct); + + r.Events.Count.ShouldBe(3); + r.ContinuationPoint.ShouldBeNull("no further MATCHING event exists"); + fake.EventStreamsCompleted.ShouldBe(1); + + // Positive control: one more matching event at the end does signal truncation. + fake.Events = events.Append(Ev(99, "Pump1")).ToArray(); + (await ds.ReadEventsAsync("Pump1", T0, T0.AddHours(1), maxEvents: 3, Ct)).ContinuationPoint.ShouldNotBeNull(); + } + + [Fact] + public async Task A_cancelled_events_read_rethrows_and_is_not_a_backend_failure() + { + var fake = new FakeHistorianGatewayClient { Events = Series(5) }; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Should.ThrowAsync(() => ds.ReadEventsAsync(null, T0, T0.AddHours(1), 3, cts.Token)); + var h = ds.GetHealthSnapshot(); + h.TotalFailures.ShouldBe(0); + h.ConsecutiveFailures.ShouldBe(0); + + // Positive control: a real fault on the same arm IS a failure. + fake.ThrowOnRead = new InvalidOperationException("boom"); + await Should.ThrowAsync(() => ds.ReadEventsAsync(null, T0, T0.AddHours(1), 3, Ct)); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(1); + } + + [Fact] + public async Task A_cancelled_native_at_time_read_rethrows_and_is_not_a_backend_failure() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + var fake = new FakeHistorianGatewayClient { ThrowOnRead = new OperationCanceledException(cts.Token) }; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, atTimeFromRaw: false); + + await Should.ThrowAsync(() => ds.ReadAtTimeAsync("T", [T0], cts.Token)); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + + // Positive control: the same arm counts a real fault. + fake.ThrowOnRead = new InvalidOperationException("boom"); + await Should.ThrowAsync(() => ds.ReadAtTimeAsync("T", [T0], Ct)); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(1); + } + + /// + /// The paged-vs-unpaged fuzz for the drain: random event series (bursts, tie clusters, foreign-source rows), random + /// windows, caps and drain budgets, paged the way the server pages a forward events read (resume at the last event's + /// time inclusive, skip the trailing ties emitted, ask cap + skip; a resumed page stalled on a tie cluster + /// over-fetches the cluster and pages within it) against a gateway-faithful fake (ascending time, ties in store + /// order, inclusive end, optionally no server-side source filter). The concatenated pages must equal the one-shot + /// uncapped read exactly — boundary ties included — and every stream must be consumed to its end whenever the + /// budget covers the overshoot. + /// + [Fact] + public async Task Paged_events_equal_the_unpaged_read_and_drain_within_budget() + { + var iterations = int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_PAGING_FUZZ_ITERATIONS"), out var n) && n > 0 + ? Math.Min(n, 4000) : 1500; + var occurred = new Dictionary(); + void Count(string k) => occurred[k] = occurred.GetValueOrDefault(k) + 1; + + for (var seed = 0; seed < iterations; seed++) + { + var rng = new Random(seed); + var store = RandomStore(rng, out var source); + var cap = rng.Next(6) switch { 0 => 1, 1 => 2, 2 => 3, 3 => rng.Next(4, 12), 4 => rng.Next(12, 60), _ => 500 }; + var budget = rng.Next(4) switch { 0 => 0, 1 => rng.Next(1, 8), _ => GatewayHistorianDataSource.DefaultEventDrainBudget }; + var (start, end) = RandomWindow(rng, store); + var tag = $"seed {seed} cap {cap} budget {budget} source {source ?? "*"} [{start:O} .. {end:O}]"; + + var truthFake = new FakeHistorianGatewayClient { EventsSource = (s, e) => Window(store, s, e) }; + var truthDs = new GatewayHistorianDataSource(truthFake, NullLogger.Instance); + var truth = (await truthDs.ReadEventsAsync(source, start, end, 0, Ct)).Events; + truth.ShouldAllBe(e => source == null || e.SourceName == source, tag); + + var fake = new FakeHistorianGatewayClient { EventsSource = (s, e) => Window(store, s, e) }; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, eventDrainBudget: budget); + var paged = await PageAsync(ds, source, start, end, cap, tag); + + paged.Select(Key).ShouldBe(truth.Select(Key), $"paged ≠ unpaged: {tag}"); + if (budget >= GatewayHistorianDataSource.DefaultEventDrainBudget) + { + fake.EventStreamsAbandoned.ShouldBe(0, $"a stream was abandoned within budget: {tag}"); + Count("drained"); + } + if (fake.EventStreamsAbandoned > 0) Count("abandoned past budget"); + if (truth.Count == 0) Count("empty"); + if (truth.Count > cap) Count("paged"); + if (truth.GroupBy(e => e.EventTimeUtc).Any(g => g.Count() > cap)) Count("tie cluster over the cap"); + if (source is not null && store.Any(e => e.SourceName != source)) Count("foreign rows"); + } + + foreach (var k in new[] { "drained", "abandoned past budget", "empty", "paged", "tie cluster over the cap", "foreign rows" }) + occurred.GetValueOrDefault(k).ShouldBeGreaterThan(0, $"the fuzz never produced '{k}' — it is vacuous for that case"); + } + + /// The server's forward events paging, against the data source (see the fuzz summary). + private static async Task> PageAsync( + IHistorianDataSource ds, string? source, DateTime start, DateTime end, int cap, string tag) + { + var all = new List(); + var skip = 0; + var resumed = false; + for (var page = 0; ; page++) + { + page.ShouldBeLessThan(10_000, $"paging must terminate: {tag}"); + var r = await ds.ReadEventsAsync(source, start, end, cap + skip, Ct); + var truncated = r.ContinuationPoint is not null; + var events = r.Events.ToList(); + if (resumed && skip > 0) + { + var i = 0; + while (i < events.Count && i < skip && events[i].EventTimeUtc == start) i++; + events.RemoveRange(0, i); + } + if (events.Count > cap) { events.RemoveRange(cap, events.Count - cap); truncated = true; } + + if (truncated && resumed && (events.Count == 0 || events[^1].EventTimeUtc == start)) + { + // Oversized tie cluster: read it whole ([T, T] on the gateway's inclusive end) and page within it. + var cluster = (await ds.ReadEventsAsync(source, start, start, 100_000, Ct)).Events + .Where(e => e.EventTimeUtc == start).ToList(); + var slice = cluster.Skip(skip).Take(cap).ToList(); + all.AddRange(slice); + if (skip + slice.Count < cluster.Count) { skip += slice.Count; continue; } + start = start.AddTicks(1); + skip = 0; + if (start > end) return all; + continue; + } + + all.AddRange(events); + if (!truncated || events.Count == 0) return all; + var boundary = events[^1].EventTimeUtc; + var ties = 0; + for (var i = events.Count - 1; i >= 0 && events[i].EventTimeUtc == boundary; i--) ties++; + // A resumed page landing on the same boundary carries its earlier skip forward. + skip = resumed && boundary == start ? skip + ties : ties; + start = boundary; + resumed = true; + } + } + + private static string Key(HistoricalEvent e) => $"{e.EventTimeUtc.Ticks}:{e.EventId}"; + + private static List RandomStore(Random rng, out string? source) + { + var count = rng.Next(5) == 0 ? 0 : rng.Next(1, 120); + var mixed = rng.Next(3) == 0; + source = rng.Next(3) == 0 ? null : "Src"; + var t = T0; + var store = new List(); + for (var i = 0; i < count; i++) + { + // Bursts: often several events share one timestamp; sometimes a big cluster. + var step = rng.Next(6) switch { 0 or 1 => 0, 2 => 1, 3 => rng.Next(1, 1000), _ => rng.Next(1, 10_000_000) }; + t = t.AddTicks(step); + var src = mixed && rng.Next(3) == 0 ? "Other" : "Src"; + store.Add(new HistorianEvent + { + Id = $"E{i:0000}", + SourceName = src, + EventTime = Timestamp.FromDateTime(t), + ReceivedTime = Timestamp.FromDateTime(t), + }); + if (rng.Next(25) == 0) + for (var k = 0; k < rng.Next(5, 40); k++) + store.Add(new HistorianEvent + { + Id = $"E{i:0000}-{k:00}", + SourceName = src, + EventTime = Timestamp.FromDateTime(t), + ReceivedTime = Timestamp.FromDateTime(t), + }); + } + return store; + } + + private static (DateTime Start, DateTime End) RandomWindow(Random rng, List store) + { + if (store.Count == 0 || rng.Next(6) == 0) return (T0, T0.AddDays(2)); + DateTime Pick() => rng.Next(3) == 0 + ? T0.AddTicks(rng.NextInt64(0, 2_000_000_000)) + : store[rng.Next(store.Count)].EventTime.ToDateTime(); + var a = Pick(); + var b = Pick(); + return a <= b ? (a, b) : (b, a); + } + + /// The gateway's ReadEvents window: ascending time, ties in store order, inclusive at both ends. + private static IEnumerable Window(List store, DateTime start, DateTime end) => + store.Where(e => e.EventTime.ToDateTime() >= start && e.EventTime.ToDateTime() <= end) + .Select((e, i) => (e, i)).OrderBy(x => x.e.EventTime.ToDateTime()).ThenBy(x => x.i).Select(x => x.e); + + private static HistorianEvent[] Series(int count, string source = "Pump1", int offsetSeconds = 0) => + Enumerable.Range(0, count).Select(i => Ev(i + offsetSeconds, source, $"E{i:000}")).ToArray(); + + private static HistorianEvent Ev(int seconds, string source, string? id = null) => new() + { + Id = id ?? $"X{seconds}", + SourceName = source, + EventTime = Timestamp.FromDateTime(T0.AddSeconds(seconds)), + ReceivedTime = Timestamp.FromDateTime(T0.AddSeconds(seconds)), + }; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs index ff441a47..f20e440b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging.Abstractions; using Shouldly; @@ -144,4 +145,24 @@ public sealed class HistorianGatewayClientAdapterTests var dataSource = Assert.IsType(GatewayHistorian.CreateDataSource(opts, services)); dataSource.MaxPreWindowRows.ShouldBe(expected); } + + [Theory] // ServerHistorian:Gateway:EventDrainBudget binds from configuration and reaches the data source + [InlineData(null, GatewayHistorianSettings.DefaultEventDrainBudget)] + [InlineData("7", 7)] + [InlineData("0", 0)] + public void Factory_passes_the_bound_event_drain_budget(string? configured, int expected) + { + var values = new Dictionary + { + ["ServerHistorian:Enabled"] = "true", + ["ServerHistorian:Endpoint"] = "https://localhost:5222", + ["ServerHistorian:ApiKey"] = "histgw_x_y", + }; + if (configured is not null) values["ServerHistorian:Gateway:EventDrainBudget"] = configured; + var opts = new Microsoft.Extensions.Configuration.ConfigurationBuilder().AddInMemoryCollection(values).Build() + .GetSection(ServerHistorianOptions.SectionName).Get()!; + using var services = new ServiceCollection().BuildServiceProvider(); + var dataSource = Assert.IsType(GatewayHistorian.CreateDataSource(opts, services)); + dataSource.EventDrainBudget.ShouldBe(expected); + } }