fix(historian): drain gateway event streams past the cap; cancellation is not a fault (lmxopcua#746)

A capped Gateway HistoryReadEvents read broke out of the gRPC stream as soon
as the page was full. Abandoning a gateway stream cancels the RPC, which the
gateway treats as a connection fault: it evicts its pooled historian session,
so the next read reconnects and concurrent reads on that session fail.

The read now keeps consuming the stream past the cap without collecting it,
bounded by ServerHistorian:Gateway:EventDrainBudget (default 10000 rows; 0 =
the old abandon-at-cap). Rows of a foreign source never count toward the
budget before the "more events" signal is seen, so an abandoned stream never
answers a page as complete. A paged-vs-unpaged fuzz (random tie series,
foreign rows, caps 1..500, budgets 0..10000) pins the page contents and the
drain; it found that abandon-before-signal case.

ReadEventsAsync and the native AtTime arm now rethrow a caller cancellation
instead of recording it as a backend failure.

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

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
     /// </summary>
     private readonly int _maxRawValuesPerRead;
 
+    /// <summary>The default of <c>eventDrainBudget</c>: <c>ServerHistorian:Gateway:EventDrainBudget</c>'s default.</summary>
+    public const int DefaultEventDrainBudget = GatewayHistorianSettings.DefaultEventDrainBudget;
+
+    /// <summary>
+    ///     Stream rows an events read drains past its cap before it abandons the gateway stream; see
+    ///     <see cref="ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/>.
+    /// </summary>
+    private readonly int _eventDrainBudget;
+
+    private long _eventStreamsAbandoned;
+
+    /// <summary>The bound events drain budget — exposed so the factory wiring is testable.</summary>
+    internal int EventDrainBudget => _eventDrainBudget;
+
+    /// <summary>Events reads that abandoned their gateway stream past the drain budget (lifetime) — test/diagnostic.</summary>
+    internal long EventStreamsAbandoned => Interlocked.Read(ref _eventStreamsAbandoned);
+
     /// <summary>A strict gateway's row ceiling, learned from its first refusal.</summary>
     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. &lt;= 0 = unbounded (the pre-#736
     ///     behaviour: an uncapped read is sent as <see cref="int.MaxValue"/>).
     /// </param>
+    /// <param name="eventDrainBudget">
+    ///     Stream rows a capped events read consumes (uncollected) past its cap before it abandons the gateway stream
+    ///     (<c>ServerHistorian:Gateway:EventDrainBudget</c>). Negative counts as 0 = abandon at the cap.
+    /// </param>
     public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> 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);
 
     /// <inheritdoc />
+    // 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<HistoricalEventsResult> 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<HistoricalEvent>(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);

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);
     }
 
     /// <summary>

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs

@@ -106,6 +106,12 @@ public sealed class ServerHistorianOptions
     /// </summary>
     public bool GatewayAtTimeFromRaw { get; init; } = true;
 
+    /// <summary>
+    ///     <c>Gateway</c> backend only: the <c>ServerHistorian:Gateway</c> sub-section — knobs that exist because of the
+    ///     HistorianGateway's wire contract (see <see cref="GatewayHistorianSettings"/>). Ignored by the other backends.
+    /// </summary>
+    public GatewayHistorianSettings Gateway { get; init; } = new();
+
     /// <summary>
     ///     Opt-in: HistoryReadEvents never reads events older than <c>now − EventReadHorizon</c> — 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;
     }
 }
+
+/// <summary>
+///     The <c>ServerHistorian:Gateway</c> sub-section: Gateway-backend knobs that exist because of the HistorianGateway's
+///     wire contract. Every default is the safe posture.
+/// </summary>
+public sealed class GatewayHistorianSettings
+{
+    /// <summary>The default of <see cref="EventDrainBudget"/>.</summary>
+    public const int DefaultEventDrainBudget = 10_000;
+
+    /// <summary>
+    ///     How many stream rows a capped HistoryReadEvents read consumes (and discards) past its page cap before it
+    ///     abandons the gateway stream. The gateway's <c>ReadEvents</c> 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).
+    /// </summary>
+    public int EventDrainBudget { get; init; } = DefaultEventDrainBudget;
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs

@@ -71,6 +71,13 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
     /// <summary>When set, ReadEvents serves <c>EventsSource(start, end)</c> instead of <see cref="Events"/>.</summary>
     public Func<DateTime, DateTime, IEnumerable<HistorianEvent>>? EventsSource;
 
+    /// <summary>Events actually pulled by the consumer across every ReadEvents stream.</summary>
+    public int EventsYieldedCount;
+    /// <summary>ReadEvents streams the consumer enumerated to the END (never broke out of — R08:1.2).</summary>
+    public int EventStreamsCompleted;
+    /// <summary>ReadEvents streams the consumer disposed before their end (an abandoned gateway stream).</summary>
+    public int EventStreamsAbandoned;
+
     /// <summary>Thrown (deferred to first enumeration) by every read method when set.</summary>
     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<WriteAck> WriteLiveValuesAsync(
@@ -272,6 +279,26 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
         RawStreamsCompleted++;
     }
 
+    private async IAsyncEnumerable<HistorianEvent> CountingEventStream(IAsyncEnumerable<HistorianEvent> 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<T> ToAsyncStream<T>(
         IReadOnlyList<T> items,
         Exception? error,

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;
+
+/// <summary>
+///     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.
+/// </summary>
+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<GatewayHistorianDataSource>.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<GatewayHistorianDataSource>.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<GatewayHistorianDataSource>.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<GatewayHistorianDataSource>.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<GatewayHistorianDataSource>.Instance);
+        using var cts = new CancellationTokenSource();
+        await cts.CancelAsync();
+
+        await Should.ThrowAsync<OperationCanceledException>(() => 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<InvalidOperationException>(() => 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<GatewayHistorianDataSource>.Instance, atTimeFromRaw: false);
+
+        await Should.ThrowAsync<OperationCanceledException>(() => 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<InvalidOperationException>(() => ds.ReadAtTimeAsync("T", [T0], Ct));
+        ds.GetHealthSnapshot().TotalFailures.ShouldBe(1);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<string, int>();
+        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<GatewayHistorianDataSource>.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<GatewayHistorianDataSource>.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");
+    }
+
+    /// <summary>The server's forward events paging, against the data source (see the fuzz summary).</summary>
+    private static async Task<List<HistoricalEvent>> PageAsync(
+        IHistorianDataSource ds, string? source, DateTime start, DateTime end, int cap, string tag)
+    {
+        var all = new List<HistoricalEvent>();
+        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<HistorianEvent> 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<HistorianEvent>();
+        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<HistorianEvent> 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);
+    }
+
+    /// <summary>The gateway's ReadEvents window: ascending time, ties in store order, inclusive at both ends.</summary>
+    private static IEnumerable<HistorianEvent> Window(List<HistorianEvent> 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)),
+    };
+}

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<GatewayHistorianDataSource>(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<string, string?>
+        {
+            ["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<ServerHistorianOptions>()!;
+        using var services = new ServiceCollection().BuildServiceProvider();
+        var dataSource = Assert.IsType<GatewayHistorianDataSource>(GatewayHistorian.CreateDataSource(opts, services));
+        dataSource.EventDrainBudget.ShouldBe(expected);
+    }
 }