fix(historian): #691.4 review — uncapped Gateway events reads no longer pre-size to int.MaxValue
GatewayHistorianDataSource.ReadEventsAsync sized its result list to maxEvents, which is int.MaxValue for NumValuesPerNode = 0: every uncapped Gateway events read (forward on master, and the new walked reverse one) threw 'Array dimensions exceeded' and counted as a backend failure. Capacity is bounded at 1024. The events walk delegate moved to ReverseEventsReader so it is tested over the real Gateway source (forward uncapped, reverse uncapped, reverse capped).
This commit is contained in:
@@ -787,6 +787,9 @@ says is complete carries no continuation point. `NumValuesPerNode == 0` stays un
|
||||
oversized-cluster over-fetch keeps exactly the events at `T` (the gateway's end is inclusive) in the same order.
|
||||
- **"More exists"** is found by walking for one event past the page (the walk has no source truncation flag to hand on).
|
||||
- **Uncapped** reverse reads are one forward read of the window, sorted and reversed.
|
||||
- ⚠️ **Also fixed (#691.4 review): uncapped events reads on Gateway.** `NumValuesPerNode = 0` reaches the source as
|
||||
`int.MaxValue`, and `GatewayHistorianDataSource` pre-sized its result list to it ("Array dimensions exceeded") — every
|
||||
uncapped Gateway events read, forward too, failed the node and counted as a backend failure. The capacity is now bounded.
|
||||
- A timestamp holding more than `MaxTieClusterOverfetch` events ⇒ **`BadTooManyOperations`** for that node.
|
||||
- ⚠️ **Gateway prerequisite:** the gateway's `RuntimeDb:EventReadMaxRows` must be `0` (its default, unlimited) or larger than
|
||||
the walk's biggest read (`2N + 18` rows, a discovery read 1,026, a tie-cluster read up to `2·MaxTieClusterOverfetch + 4`).
|
||||
|
||||
+2
-1
@@ -425,7 +425,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
try
|
||||
{
|
||||
var hasCap = maxEvents > 0;
|
||||
var collected = new List<HistoricalEvent>(hasCap ? maxEvents : 0);
|
||||
// Bounded initial capacity: an uncapped read arrives as int.MaxValue ("Array dimensions exceeded", #691.4 review).
|
||||
var collected = new List<HistoricalEvent>(hasCap ? Math.Min(maxEvents, 1024) : 0);
|
||||
var truncated = false;
|
||||
|
||||
await foreach (var wireEvent in _client
|
||||
|
||||
@@ -5311,22 +5311,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
// Walk for one more than the fetch: finding it is the "more events exist" signal a source gives with its
|
||||
// continuation point (the walk itself cannot flag truncation).
|
||||
var want = numValuesPerNode == 0 ? 0u : (uint)Math.Min(uint.MaxValue - 1L, (long)fetch + 1);
|
||||
var budget = Math.Max(1, MaxTieClusterOverfetch);
|
||||
var walkSource = sourceName;
|
||||
var (rows, next) = await ReverseWalk.WalkAsync<HistoricalEvent>(
|
||||
async (s, e, cap, token) =>
|
||||
{
|
||||
var r = await source.ReadEventsAsync(walkSource, s, e, cap == 0 ? int.MaxValue : cap, reverse: false, token)
|
||||
.ConfigureAwait(false);
|
||||
return new ForwardSegmentRead<HistoricalEvent>(r.Events, r.ContinuationPoint is not null);
|
||||
},
|
||||
ev => ev.EventTimeUtc.Ticks, EventIdTieOrder.Instance, startUtc, endUtc, want, budget,
|
||||
source.TimestampResolutionTicks, Clock.GetUtcNow().UtcDateTime, walkHint,
|
||||
at => new HistoryReadLimitExceededException(budget,
|
||||
$"A reverse events read of '{walkSource}' at {at:O} found more than {budget} events at one " +
|
||||
"timestamp; the backend reads forward only, so the latest of them cannot be selected. Increase " +
|
||||
"ServerHistorian:MaxTieClusterOverfetch."),
|
||||
ct)
|
||||
var (rows, next) = await ReverseEventsReader.ReadAsync(
|
||||
source, sourceName, startUtc, endUtc, want, MaxTieClusterOverfetch, Clock.GetUtcNow().UtcDateTime,
|
||||
walkHint, ct)
|
||||
.ConfigureAwait(false);
|
||||
walkHint = next;
|
||||
var more = want > 0 && rows.Count > fetch;
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
|
||||
/// <summary>One walked-back reverse events page (<see cref="ReverseEventsReader"/>).</summary>
|
||||
/// <param name="Events">At most the requested count, newest first by <c>(EventTimeUtc, EventId)</c>.</param>
|
||||
/// <param name="Hint">What the next page of the same read resumes with.</param>
|
||||
internal readonly record struct ReverseEventsPage(List<HistoricalEvent> Events, ReverseWalkHint Hint);
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #691.4: serves a Part 11 reverse events read — <c>(lo, hi]</c> newest-first — on a backend that reads events
|
||||
/// forward only (<see cref="IHistorianDataSource.SupportsReverseEvents"/> = <c>false</c>: Gateway, Null) with the
|
||||
/// <see cref="ReverseWalk"/>. A segment read is cut short when the source flags truncation (its continuation point); ties
|
||||
/// are ordered by <see cref="EventIdTieOrder"/> because the gateway orders by event time only.
|
||||
/// </summary>
|
||||
internal static class ReverseEventsReader
|
||||
{
|
||||
/// <summary>Reads <c>(lo, hi]</c> newest-first from forward event reads.</summary>
|
||||
/// <param name="source">A backend that reads events forward.</param>
|
||||
/// <param name="sourceName">Event-source filter (null = all).</param>
|
||||
/// <param name="hi">The later (inclusive) bound.</param>
|
||||
/// <param name="lo">The earlier (exclusive) bound.</param>
|
||||
/// <param name="want">How many events; <c>0</c> = all in the window (one forward read).</param>
|
||||
/// <param name="maxTieCluster">The tie budget (<c>MaxTieClusterOverfetch</c>) — a ceiling only.</param>
|
||||
/// <param name="nowUtc">Now, for the look-ahead clamp.</param>
|
||||
/// <param name="hint">What the previous page carried (default = none).</param>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <returns>The events, newest first, and the hint for the next page.</returns>
|
||||
/// <exception cref="HistoryReadLimitExceededException">One timestamp holds more than the tie budget.</exception>
|
||||
public static async Task<ReverseEventsPage> ReadAsync(
|
||||
IHistorianDataSource source, string? sourceName, DateTime hi, DateTime lo, uint want, int maxTieCluster,
|
||||
DateTime nowUtc, ReverseWalkHint hint, CancellationToken ct)
|
||||
{
|
||||
var budget = Math.Max(1, maxTieCluster);
|
||||
var (rows, next) = await ReverseWalk.WalkAsync<HistoricalEvent>(
|
||||
async (start, end, cap, token) =>
|
||||
{
|
||||
// cap 0 = the whole window: int.MaxValue is the events contract's "no cap" (a non-positive value would
|
||||
// mean the backend's DEFAULT cap).
|
||||
var r = await source.ReadEventsAsync(sourceName, start, end, cap == 0 ? int.MaxValue : cap, reverse: false, token)
|
||||
.ConfigureAwait(false);
|
||||
return new ForwardSegmentRead<HistoricalEvent>(r.Events, r.ContinuationPoint is not null);
|
||||
},
|
||||
ev => ev.EventTimeUtc.Ticks, EventIdTieOrder.Instance, hi, lo, want, budget, source.TimestampResolutionTicks,
|
||||
nowUtc, hint,
|
||||
at => new HistoryReadLimitExceededException(budget,
|
||||
$"A reverse events read of '{sourceName}' at {at:O} found more than {budget} events at one timestamp; the " +
|
||||
"backend reads forward only, so the latest of them cannot be selected. Increase " +
|
||||
"ServerHistorian:MaxTieClusterOverfetch."),
|
||||
ct)
|
||||
.ConfigureAwait(false);
|
||||
return new ReverseEventsPage(rows, next);
|
||||
}
|
||||
}
|
||||
+4
-1
@@ -54,6 +54,9 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
|
||||
public int LastReadEventsMaxEvents;
|
||||
public int ReadEventsCallCount;
|
||||
|
||||
/// <summary>When set, ReadEvents serves <c>EventsSource(start, end)</c> instead of <see cref="Events"/>.</summary>
|
||||
public Func<DateTime, DateTime, IEnumerable<HistorianEvent>>? EventsSource;
|
||||
|
||||
/// <summary>Thrown (deferred to first enumeration) by every read method when set.</summary>
|
||||
public Exception? ThrowOnRead;
|
||||
|
||||
@@ -187,7 +190,7 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
|
||||
LastReadEventsEndUtc = endUtc;
|
||||
LastReadEventsMaxEvents = maxEvents;
|
||||
ReadEventsCallCount++;
|
||||
return ToAsyncStream(Events, ThrowOnRead, ct);
|
||||
return ToAsyncStream(EventsSource is null ? Events : EventsSource(startUtc, endUtc).ToList(), ThrowOnRead, ct);
|
||||
}
|
||||
|
||||
public Task<WriteAck> WriteLiveValuesAsync(
|
||||
|
||||
+76
@@ -0,0 +1,76 @@
|
||||
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;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #691.4 review: events read uncapped (and walked back newest-first) through the REAL
|
||||
/// <see cref="GatewayHistorianDataSource"/> over a gateway-shaped fake — inclusive <c>[start, end]</c>, ordered by event time.
|
||||
/// </summary>
|
||||
public sealed class GatewayReverseEventsTests
|
||||
{
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
|
||||
private static readonly DateTime T0 = new(2026, 10, 12, 18, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private static FakeHistorianGatewayClient Store(int count)
|
||||
{
|
||||
var stored = Enumerable.Range(0, count)
|
||||
.Select(i => new HistorianEvent { Id = $"e{i:0000}", Type = "Alarm", EventTime = Timestamp.FromDateTime(T0.AddSeconds(i)) })
|
||||
.ToList();
|
||||
return new FakeHistorianGatewayClient
|
||||
{
|
||||
EventsSource = (start, end) => stored.Where(e => e.EventTime.ToDateTime() >= start && e.EventTime.ToDateTime() <= end),
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <c>NumValuesPerNode = 0</c> reaches the source as <c>int.MaxValue</c>; the result list used to be pre-sized to it
|
||||
/// ("Array dimensions exceeded") — every uncapped Gateway events read failed and counted as a backend failure.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task An_uncapped_forward_events_read_is_served()
|
||||
{
|
||||
var fake = Store(30);
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
|
||||
var r = await ds.ReadEventsAsync(null, T0, T0.AddMinutes(1), int.MaxValue, Ct);
|
||||
|
||||
r.Events.Count.ShouldBe(30);
|
||||
r.ContinuationPoint.ShouldBeNull();
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_uncapped_reverse_events_read_is_one_forward_read_newest_first()
|
||||
{
|
||||
var fake = Store(30);
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
|
||||
var page = await ReverseEventsReader.ReadAsync(ds, null, hi: T0.AddSeconds(25), lo: T0.AddSeconds(5), want: 0,
|
||||
maxTieCluster: 65536, nowUtc: T0.AddHours(1), hint: default, Ct);
|
||||
|
||||
page.Events.Select(e => e.EventId).ShouldBe(Enumerable.Range(6, 20).Reverse().Select(i => $"e{i:0000}"));
|
||||
fake.ReadEventsCallCount.ShouldBe(1);
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_capped_reverse_events_read_walked_back_over_the_gateway_returns_the_latest_newest_first()
|
||||
{
|
||||
var fake = Store(300);
|
||||
var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
|
||||
var page = await ReverseEventsReader.ReadAsync(ds, null, hi: T0.AddSeconds(200), lo: T0.AddSeconds(50), want: 10,
|
||||
maxTieCluster: 65536, nowUtc: T0.AddHours(1), hint: default, Ct);
|
||||
|
||||
page.Events.Select(e => e.EventId).ShouldBe(Enumerable.Range(191, 10).Reverse().Select(i => $"e{i:0000}"));
|
||||
fake.LastReadEventsStartUtc.ShouldBeLessThanOrEqualTo(fake.LastReadEventsEndUtc);
|
||||
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user