fix(historian): respect gateway ceilings and history value contracts
This commit is contained in:
@@ -227,7 +227,27 @@ internal static class GatewayAtTimeReader
|
||||
{
|
||||
if (!beforeKnown)
|
||||
{
|
||||
before = await WalkBackAsync(ctx, qs, walkKnown, walkWidth).ConfigureAwait(false);
|
||||
// #696: prove the future stretch empty before jumping to now. Historian clocks or imports
|
||||
// can place real rows there; a full bounded probe seeds the original backward search.
|
||||
var searchEnd = qs;
|
||||
if (qs > now && walkKnown is null)
|
||||
{
|
||||
var future = await ReadTrustedAsync(ctx, MsFloor(now),
|
||||
Min(MsFloor(qs) + OneMs, horizon), ReadRows).ConfigureAwait(false);
|
||||
StoredSample? latest = null;
|
||||
foreach (var row in future.Rows)
|
||||
if (row.TimeUtc <= qs) latest = row;
|
||||
if (future.Complete)
|
||||
{
|
||||
before = latest;
|
||||
beforeKnown = latest is not null;
|
||||
searchEnd = now;
|
||||
}
|
||||
else if (latest is { } found)
|
||||
walkKnown = found.TimeUtc;
|
||||
}
|
||||
if (!beforeKnown)
|
||||
before = await WalkBackAsync(ctx, searchEnd, walkKnown, walkWidth).ConfigureAwait(false);
|
||||
beforeKnown = true;
|
||||
}
|
||||
prev = before;
|
||||
|
||||
+24
-1
@@ -152,6 +152,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -239,7 +240,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++)
|
||||
{
|
||||
var budget = (int)Math.Min(int.MaxValue, (long)cap + room + RawAggregateReader.PreWindowAllowance);
|
||||
if (_ceiling.Learned is var known and > 0 && budget > known) break; // over a strict ceiling: refused anyway
|
||||
// #695: doubling can leap over the ceiling while the cluster still fits. Spend the remaining room once.
|
||||
if (_ceiling.Learned is var known and > 0 && budget > known)
|
||||
{
|
||||
var available = known - cap - RawAggregateReader.PreWindowAllowance;
|
||||
if (available <= 0) break;
|
||||
room = Math.Min(room, available);
|
||||
budget = cap + room + RawAggregateReader.PreWindowAllowance;
|
||||
}
|
||||
if (budget <= pass.Budget) break; // no more room, never retry the same full read
|
||||
pass = await PassAsync(budget).ConfigureAwait(false);
|
||||
if (pass.Result.Complete) return pass.Result.Samples;
|
||||
preWindow = Math.Max(preWindow, pass.Result.PreWindowRows);
|
||||
@@ -365,6 +374,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -408,6 +418,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -466,6 +477,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -486,6 +498,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -581,6 +594,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordReadFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -655,6 +669,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>Classifies actual gRPC/package transport failures without treating refusals as outages (#697).</summary>
|
||||
private static bool IsTransportFailure(Exception ex) => ex switch
|
||||
{
|
||||
ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayUnavailableException => true,
|
||||
RpcException rpc => rpc.StatusCode is Grpc.Core.StatusCode.Unavailable or Grpc.Core.StatusCode.DeadlineExceeded,
|
||||
TimeoutException => true,
|
||||
_ => ex.InnerException is { } inner && IsTransportFailure(inner),
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Records a failed read: bumps the health counters and logs a generic, redaction-safe
|
||||
/// debug line (no tag, host, or value). The exception itself is rethrown by the caller.
|
||||
|
||||
@@ -137,6 +137,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -232,6 +233,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -316,6 +318,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -443,6 +446,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -538,6 +542,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -701,6 +706,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
|
||||
private void RecordSuccess() => _health.RecordSuccess();
|
||||
|
||||
/// <summary>Classifies provider transport/timeouts, keeping SQL/query errors distinct (#697).</summary>
|
||||
private static bool IsTransportFailure(Exception ex) => ex is NpgsqlException { IsTransient: true } and not PostgresException
|
||||
|| ex is PostgresException pg && (pg.SqlState.StartsWith("08", StringComparison.Ordinal)
|
||||
|| pg.SqlState is "57P01" or "57P02" or "57P03");
|
||||
|
||||
private void RecordFailure(Exception ex)
|
||||
{
|
||||
// A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).
|
||||
|
||||
+8
@@ -132,6 +132,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -229,6 +230,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex) when (ex is not NotSupportedException)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -306,6 +308,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -401,6 +404,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -512,6 +516,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -635,6 +640,9 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
|
||||
private void RecordSuccess() => _health.RecordSuccess();
|
||||
|
||||
/// <summary>Classifies provider transport/timeouts, keeping SQL/query errors distinct (#697).</summary>
|
||||
private static bool IsTransportFailure(Exception ex) => ex is SqlException sql && sql.Number is -2 or 0 or 53 or 64 or 233 or 258 or 10053 or 10054 or 10060 or 10061 or 10065 or 11001 or 40197 or 40613;
|
||||
|
||||
private void RecordFailure(Exception ex)
|
||||
{
|
||||
// A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).
|
||||
|
||||
@@ -5129,6 +5129,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
|
||||
RunBounded(work, HistoryReadBatchParallelism);
|
||||
});
|
||||
NormalizeHistoryValues(nodesToProcess, results);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -5225,6 +5226,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
|
||||
RunBounded(work, HistoryReadBatchParallelism);
|
||||
});
|
||||
NormalizeHistoryValues(nodesToProcess, results);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -5516,8 +5518,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
// One node's backend failure must not throw out of the batch — surface Bad for THIS node
|
||||
// only. Logged via the injected ILogger.
|
||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryReadEvents failed for node {NodeId}", handle.NodeId);
|
||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
|
||||
errors[handle.Index] = HistoryReadFailureStatus(ex);
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6233,7 +6235,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
// Bad status for THIS node only. Logged through the injected ILogger
|
||||
// rather than swallowed silently.
|
||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead failed for node {NodeId}", handle.NodeId);
|
||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
||||
errors[handle.Index] = HistoryReadFailureStatus(ex);
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6722,7 +6725,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
{
|
||||
// One node's backend failure must not throw out of the batch — Bad for THIS node only.
|
||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryReadRaw (paged) failed for node {NodeId}", handle.NodeId);
|
||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
||||
errors[handle.Index] = HistoryReadFailureStatus(ex);
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6829,8 +6833,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead (modified) failed for node {NodeId}", handle.NodeId);
|
||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
|
||||
errors[handle.Index] = HistoryReadFailureStatus(ex);
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = HistoryReadFailureStatus(ex) };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6879,6 +6883,49 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Transport failures are retryable; unsupported operations retain their distinct status (#697).</summary>
|
||||
private static uint HistoryReadFailureStatus(Exception error)
|
||||
{
|
||||
for (Exception? ex = error; ex is not null; ex = ex.InnerException)
|
||||
if (ex is System.IO.IOException or System.Net.Sockets.SocketException or System.Net.Http.HttpRequestException or TimeoutException)
|
||||
return StatusCodes.BadNoCommunication;
|
||||
return StatusCodes.BadHistoryOperationUnsupported;
|
||||
}
|
||||
|
||||
/// <summary>Normalize stored and AtTime history to each scalar node's declared DataType (#698).</summary>
|
||||
private void NormalizeHistoryValues(List<NodeHandle> handles, IList<SdkHistoryReadResult> results)
|
||||
{
|
||||
foreach (var handle in handles)
|
||||
{
|
||||
if (AtTimeValueType(handle.NodeId) is not { } type || results[handle.Index]?.HistoryData is not { } body) continue;
|
||||
var values = ExtensionObject.ToEncodeable(body) switch
|
||||
{
|
||||
HistoryData data => data.DataValues,
|
||||
_ => null,
|
||||
};
|
||||
if (values is null) continue;
|
||||
foreach (var value in values)
|
||||
{
|
||||
if (value.Value is null) continue;
|
||||
try
|
||||
{
|
||||
// A gateway number cannot reconstruct a DateTime or a String which was never stored.
|
||||
if ((type == TypeCode.DateTime && value.Value is not DateTime)
|
||||
|| (type == TypeCode.String && value.Value is not string)) throw new InvalidCastException();
|
||||
var converted = Convert.ChangeType(value.Value, type, System.Globalization.CultureInfo.InvariantCulture);
|
||||
if (converted is float f && float.IsInfinity(f) && value.Value is double d && double.IsFinite(d))
|
||||
throw new OverflowException();
|
||||
value.WrappedValue = new Variant(converted);
|
||||
}
|
||||
catch (Exception ex) when (ex is InvalidCastException or OverflowException or FormatException)
|
||||
{
|
||||
value.WrappedValue = Variant.Null;
|
||||
value.StatusCode = StatusCodes.BadTypeMismatch;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Project a plain sample list into an SDK <see cref="HistoryData"/> (the paged Raw path
|
||||
/// works on a trimmed <see cref="IReadOnlyList{T}"/> rather than a whole <see cref="HistorianRead"/>).</summary>
|
||||
/// <param name="samples">The samples to project (already trimmed of boundary duplicates).</param>
|
||||
|
||||
@@ -100,16 +100,22 @@ internal static class ReverseRawReader
|
||||
{
|
||||
var asked = limit > 0 && cap > limit ? limit : cap; // cap 0 (the whole window) is never cut
|
||||
IReadOnlyList<DataValueSnapshot> samples;
|
||||
try
|
||||
for (var attempt = 0; ; attempt++)
|
||||
{
|
||||
samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
|
||||
}
|
||||
catch (HistoryReadLimitExceededException ex) when (asked > 0 && ex.PreWindowRows > 0 && asked - ex.PreWindowRows >= 1)
|
||||
{
|
||||
// #694: rows just before the segment crowded the backend's budget — read it smaller; a full answer
|
||||
// then counts as truncated (the walk narrows), never as complete.
|
||||
asked -= ex.PreWindowRows;
|
||||
samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
|
||||
try
|
||||
{
|
||||
samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
|
||||
break;
|
||||
}
|
||||
catch (HistoryReadLimitExceededException ex) when (asked > 0 && ex.PreWindowRows > 0 && attempt < 2)
|
||||
{
|
||||
// #695: the discovery request may be smaller than the pre-window cluster. Size against
|
||||
// the ceiling, not that request; a second pass can discover more crowding rows.
|
||||
var ceilingRoom = source.MaxRawValuesPerRead is var learned and > 0 ? learned : asked;
|
||||
var smaller = Math.Min(asked - 1, ceilingRoom - ex.PreWindowRows - 1);
|
||||
if (smaller < 1) throw;
|
||||
asked = smaller;
|
||||
}
|
||||
}
|
||||
return new ForwardSegmentRead<DataValueSnapshot>(samples, asked > 0 && samples.Count >= asked);
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user