Files
lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs
T
Joseph Doherty 6669fae7e5 feat(historian): ServerHistorian:MaxRawValuesPerRead caps raw pages and backend reads
Review R08:1.1: an uncapped HistoryRead-Raw (NumValuesPerNode = 0) read a
tag's whole window into memory on every backend (SQL TOP(int.MaxValue),
QuestDB LIMIT, Gateway max_values = int.MaxValue + a second snapshot copy).

New ServerHistorian:MaxRawValuesPerRead (default 10 000; <= 0 unbounded with
a Validate() warning; 1 raised to 2). Each backend data source takes it,
reports it as IHistorianDataSource.MaxRawValuesPerRead (the Gateway: min
with a strict gateway's learned ceiling), so the node manager's existing
RawPageLimit paging serves N = 0 and N > cap as pages of at most the cap
with a continuation point. Independently, a backend raw read asking for 0
or more than the cap reads cap + 1 rows and is REFUSED
(HistoryReadLimitExceededException => BadTooManyOperations) when the window
holds more - never answered truncated, since the server's paging reads a
short answer as the last page. Bound through the Host registry, the
SqlServer/QuestDb providers and GatewayHistorian.CreateDataSource.

Tests: options (default/binding/warnings), provider binding, Gateway unit +
through a real SDK node manager (uncapped -> 999/998/998/605 + CPs), and a
conformance case for the SQL backends.

Refs #736

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 06:29:10 -04:00

143 lines
6.3 KiB
C#

using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
using System.Collections.Frozen;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// <summary>
/// <see cref="IHistorianBackendProvider"/> for the QuestDB historian. Supports every
/// <see cref="HistorianValueKind"/> and HistoryUpdate (soft delete). All seams share ONE connection helper, ONE
/// schema bootstrapper, ONE tag catalog (R7: the only "known tag" authority) and ONE WAL visibility wait, so the
/// schema is ensured once and known tags are cached once.
/// </summary>
/// <remarks>
/// The shared state is built on the first <c>Create*</c> call, not in the constructor: the constructor does no
/// validation and no I/O, so resolving the provider for a backend nobody consumes never fails on its options
/// (they are validated, consumer-gated, by the host's <c>ServerHistorianOptionsValidator</c>). The first
/// <c>Create*</c> call's service provider supplies the logger factory the shared pieces log through.
/// <para>The provider owns the shared connection pool and disposes it (IAsyncDisposable/IDisposable); the host
/// registers the provider through a DI factory so the container disposes it at shutdown.</para>
/// </remarks>
public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsyncDisposable, IDisposable
{
private static readonly FrozenSet<HistorianValueKind> Kinds =
Enum.GetValues<HistorianValueKind>().ToFrozenSet();
private readonly QuestDbHistorianOptions _options;
private readonly object _initLock = new();
private Shared? _shared;
private bool _disposed;
/// <summary>Creates the provider over the <c>ServerHistorian:QuestDb</c> options (validated elsewhere).</summary>
/// <param name="options">The <c>ServerHistorian:QuestDb</c> options.</param>
/// <param name="maxTimeWeightedRawRows">The shared <c>ServerHistorian:MaxTimeWeightedRawRows</c> cap (&lt;= 0 = unbounded).</param>
/// <param name="maxRawValuesPerRead">The shared <c>ServerHistorian:MaxRawValuesPerRead</c> raw cap (&lt;= 0 = unbounded; R08:1.1).</param>
public QuestDbHistorianProvider(QuestDbHistorianOptions options,
int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, int maxRawValuesPerRead = 0)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
_maxTimeWeightedRawRows = maxTimeWeightedRawRows;
_maxRawValuesPerRead = maxRawValuesPerRead;
}
private readonly int _maxTimeWeightedRawRows;
private readonly int _maxRawValuesPerRead;
/// <inheritdoc />
public string Name => "QuestDb";
/// <inheritdoc />
public IReadOnlySet<HistorianValueKind> SupportedKinds => Kinds;
private static ILoggerFactory Loggers(IServiceProvider sp) =>
(sp.GetService(typeof(ILoggerFactory)) as ILoggerFactory) ?? NullLoggerFactory.Instance;
private Shared Get(IServiceProvider sp)
{
if (Volatile.Read(ref _shared) is { } built) return built;
lock (_initLock)
{
if (_shared is { } raced) return raced;
var loggers = Loggers(sp);
ObjectDisposedException.ThrowIf(_disposed, this);
var conn = new QuestDbConnection(_options, loggers.CreateLogger<QuestDbConnection>());
var bootstrap = new QuestDbSchemaBootstrapper(conn, _options, loggers.CreateLogger<QuestDbSchemaBootstrapper>());
var wal = new QuestDbWalVisibility(conn, _options.VisibilityTimeout, loggers.CreateLogger<QuestDbWalVisibility>());
var shared = new Shared(conn, bootstrap, new QuestDbTagCatalog(conn, bootstrap, wal), wal);
Volatile.Write(ref _shared, shared);
return shared;
}
}
/// <inheritdoc />
public IHistorianDataSource CreateDataSource(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog,
Loggers(sp).CreateLogger<QuestDbHistorianDataSource>(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows,
maxRawValuesPerRead: _maxRawValuesPerRead);
}
/// <inheritdoc />
public IHistorianValueWriter CreateValueWriter(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianValueWriter(s.Conn, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbHistorianValueWriter>());
}
/// <inheritdoc />
public IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbAlarmHistorianWriter(s.Conn, s.Bootstrap, s.Wal,
Loggers(sp).CreateLogger<QuestDbAlarmHistorianWriter>());
}
/// <inheritdoc />
public IHistorianProvisioning CreateProvisioning(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbTagProvisioner(s.Conn, s.Bootstrap, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbTagProvisioner>());
}
/// <inheritdoc />
// The soft-delete updater: deletes write tombstones (INSERT only, never QuestDB UPDATE/DELETE).
public IHistorianDataUpdater? CreateUpdater(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianDataUpdater(s.Conn, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbHistorianDataUpdater>());
}
/// <summary>Disposes the shared connection pool, when one was built. Idempotent.</summary>
/// <returns>A task completing when the pool is disposed.</returns>
public async ValueTask DisposeAsync()
{
if (TakeForDispose() is { } conn) await conn.DisposeAsync().ConfigureAwait(false);
}
/// <summary>Disposes the shared connection pool, when one was built. Idempotent.</summary>
public void Dispose() => TakeForDispose()?.Dispose();
private QuestDbConnection? TakeForDispose()
{
lock (_initLock)
{
if (_disposed) return null;
_disposed = true;
var conn = _shared?.Conn;
_shared = null;
return conn;
}
}
private sealed record Shared(
QuestDbConnection Conn, QuestDbSchemaBootstrapper Bootstrap, QuestDbTagCatalog Catalog, QuestDbWalVisibility Wal);
}