fix(historian): page by the backend's timestamp resolution; QuestDB AtTime stamps the requested time
IHistorianDataSource.TimestampResolutionTicks (default 1; QuestDB 10). A drained tie cluster now steps off T by the resolution, so a forward one-value-per-page read on a microsecond backend no longer re-reads the same sample forever (C1). QuestDB AtTime results carry the client's original request time (M3); regression for the NEXT seek skipping a tombstone (M1). Conformance backend wires the real writer and updater. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
@@ -345,6 +345,51 @@ public sealed class HistoryPagingTests
|
||||
store.TryTake(null, cp3).ShouldNotBeNull(); // retained
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SliceTieCluster_forward_drain_steps_by_the_backend_resolution()
|
||||
{
|
||||
// A microsecond backend (QuestDB, 10 ticks) truncates T + 1 tick back to T, so the drained cluster must be
|
||||
// stepped off by the resolution, not by one tick, or the same sample comes back forever.
|
||||
var t = new DateTime(2026, 1, 1, 0, 0, 5, DateTimeKind.Utc);
|
||||
|
||||
HistoryPaging.SliceTieCluster(
|
||||
clusterCount: 1, skip: 1, cap: 1, boundaryT: t, endUtc: t.AddHours(1),
|
||||
out _, out var sliceCount, out var nextStartUtc, out var nextSkip, resolutionTicks: 10);
|
||||
|
||||
sliceCount.ShouldBe(0);
|
||||
nextStartUtc.ShouldBe(t.AddTicks(10));
|
||||
nextSkip.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SliceTieCluster_reverse_drain_steps_down_by_the_backend_resolution()
|
||||
{
|
||||
var t = new DateTime(2026, 1, 1, 0, 0, 5, DateTimeKind.Utc);
|
||||
|
||||
HistoryPaging.SliceTieCluster(
|
||||
clusterCount: 1, skip: 1, cap: 1, boundaryT: t, endUtc: t.AddHours(-1),
|
||||
out _, out _, out var nextStartUtc, out _, reverse: true, resolutionTicks: 10);
|
||||
|
||||
nextStartUtc.ShouldBe(t.AddTicks(-10));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SliceTieCluster_forward_drain_terminates_when_one_resolution_step_passes_the_end()
|
||||
{
|
||||
var t = new DateTime(2026, 1, 1, 0, 0, 5, DateTimeKind.Utc);
|
||||
|
||||
HistoryPaging.SliceTieCluster(
|
||||
clusterCount: 1, skip: 1, cap: 1, boundaryT: t, endUtc: t.AddTicks(9),
|
||||
out _, out _, out var nextStartUtc, out _, resolutionTicks: 10);
|
||||
|
||||
nextStartUtc.ShouldBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SliceTieCluster_rejects_a_resolution_below_one() =>
|
||||
Should.Throw<ArgumentOutOfRangeException>(() => HistoryPaging.SliceTieCluster(
|
||||
1, 1, 1, DateTime.UtcNow, DateTime.UtcNow.AddHours(1), out _, out _, out _, out _, resolutionTicks: 0));
|
||||
|
||||
private static HistoryContinuationState RawState() => new(
|
||||
"WW.Tag", DateTime.UtcNow, DateTime.UtcNow.AddHours(1),
|
||||
BoundarySkipCount: 1, NumValuesPerNode: 50);
|
||||
|
||||
@@ -357,6 +357,56 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
|
||||
await host.DisposeAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Forward paging at <c>NumValuesPerNode = 1</c> over a MICROSECOND backend (QuestDB: it truncates every inbound
|
||||
/// time to 10 ticks and reports <c>TimestampResolutionTicks = 10</c>) returns every sample exactly once and
|
||||
/// terminates. Before the fix the drained one-sample "cluster" resumed at <c>T + 1 tick</c>, which the backend
|
||||
/// truncated back to <c>T</c>: the same sample came back with no boundary skip, and the read looped forever.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Raw_forward_read_with_one_value_per_page_on_a_microsecond_backend_returns_every_sample_once()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
var nm = server.NodeManager!;
|
||||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||||
// Microsecond-spaced samples, so a one-tick step would land strictly between two of them.
|
||||
var series = Enumerable.Range(0, 6)
|
||||
.Select(i => { var t = Epoch.AddTicks(10L * i); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
|
||||
.ToArray();
|
||||
nm.HistorianDataSource = new MicrosecondHistorianDataSource(series);
|
||||
|
||||
nm.EnsureVariable("eq-1/us", parentFolderNodeId: null, displayName: "Us", dataType: "Double",
|
||||
writable: false, historianTagname: "WW.Us", realm: AddressSpaceRealm.Uns);
|
||||
var nodeId = nm.TryGetVariable("eq-1/us")!.NodeId;
|
||||
|
||||
var collected = await DrainAsync(nm, nodeId, start: Epoch, end: Epoch.AddHours(1), max: 1);
|
||||
|
||||
collected.ShouldBe([0d, 1d, 2d, 3d, 4d, 5d]);
|
||||
await host.DisposeAsync();
|
||||
}
|
||||
|
||||
/// <summary>The reverse mirror on the same microsecond backend: newest first, each sample once.</summary>
|
||||
[Fact]
|
||||
public async Task Raw_reverse_read_with_one_value_per_page_on_a_microsecond_backend_returns_every_sample_once()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
var nm = server.NodeManager!;
|
||||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||||
var series = Enumerable.Range(0, 6)
|
||||
.Select(i => { var t = Epoch.AddTicks(10L * (i + 1)); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
|
||||
.ToArray();
|
||||
nm.HistorianDataSource = new MicrosecondHistorianDataSource(series);
|
||||
|
||||
nm.EnsureVariable("eq-1/usrev", parentFolderNodeId: null, displayName: "UsRev", dataType: "Double",
|
||||
writable: false, historianTagname: "WW.UsRev", realm: AddressSpaceRealm.Uns);
|
||||
var nodeId = nm.TryGetVariable("eq-1/usrev")!.NodeId;
|
||||
|
||||
var collected = await DrainAsync(nm, nodeId, start: Epoch.AddHours(1), end: Epoch, max: 1);
|
||||
|
||||
collected.ShouldBe([5d, 4d, 3d, 2d, 1d, 0d]);
|
||||
await host.DisposeAsync();
|
||||
}
|
||||
|
||||
/// <summary>Pages a Raw read to completion and returns the union of the pages in arrival order.</summary>
|
||||
private static Task<List<double>> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max)
|
||||
{
|
||||
@@ -520,6 +570,43 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
|
||||
public void Dispose() { }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A microsecond-resolution backend, shaped like QuestDB: every inbound window bound is TRUNCATED to 10 ticks
|
||||
/// (as <c>QuestDbConnection.Round</c> does) before the series is filtered, and the source reports
|
||||
/// <see cref="TimestampResolutionTicks"/> = 10.
|
||||
/// </summary>
|
||||
private sealed class MicrosecondHistorianDataSource(IReadOnlyList<DataValueSnapshot> series) : IHistorianDataSource
|
||||
{
|
||||
private readonly SeriesHistorianDataSource _inner = new(series);
|
||||
|
||||
public long TimestampResolutionTicks => 10;
|
||||
|
||||
private static DateTime Truncate(DateTime t) => new(t.Ticks - t.Ticks % 10, DateTimeKind.Utc);
|
||||
|
||||
public Task<HistorianRead> ReadRawAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||||
CancellationToken cancellationToken) =>
|
||||
_inner.ReadRawAsync(fullReference, Truncate(startUtc), Truncate(endUtc), maxValuesPerNode, cancellationToken);
|
||||
|
||||
public Task<HistorianRead> ReadProcessedAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||||
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
|
||||
_inner.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);
|
||||
|
||||
public Task<HistorianRead> ReadAtTimeAsync(
|
||||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
||||
_inner.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
|
||||
|
||||
public Task<HistoricalEventsResult> ReadEventsAsync(
|
||||
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
||||
CancellationToken cancellationToken) =>
|
||||
_inner.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);
|
||||
|
||||
public HistorianHealthSnapshot GetHealthSnapshot() => _inner.GetHealthSnapshot();
|
||||
|
||||
public void Dispose() { }
|
||||
}
|
||||
|
||||
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
|
||||
{
|
||||
var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}";
|
||||
|
||||
Reference in New Issue
Block a user