fix(historian): respect gateway ceilings and history value contracts
This commit is contained in:
+15
-5
@@ -573,7 +573,8 @@ node without `isHistorized` has no HistoryRead access bit and answers
|
|||||||
| Historized node, historian configured, time range empty | `GoodNoData` |
|
| Historized node, historian configured, time range empty | `GoodNoData` |
|
||||||
| Historized node, historian NOT configured (`Enabled=false` / Null source) | `GoodNoData` (empty) for Raw / AtTime; Processed ⇒ `BadAggregateNotSupported` (#682) |
|
| Historized node, historian NOT configured (`Enabled=false` / Null source) | `GoodNoData` (empty) for Raw / AtTime; Processed ⇒ `BadAggregateNotSupported` (#682) |
|
||||||
| Non-historized node | `BadHistoryOperationUnsupported` |
|
| Non-historized node | `BadHistoryOperationUnsupported` |
|
||||||
| Backend timeout or exception | `BadHistoryOperationUnsupported` per node; other nodes in the same batch are unaffected |
|
| Backend transport outage (#697) | `BadNoCommunication` per node across Raw, Modified, Processed, AtTime and Events; cancelled/deadline-limited server reads retain `BadTimeout` |
|
||||||
|
| Other backend exception / unsupported operation | `BadHistoryOperationUnsupported` per node; other nodes in the same batch are unaffected |
|
||||||
| `Backend=SqlServer`, `AutoCreateSchema=false`, schema missing/behind | every read/write fails with a clear error naming the missing schema version (health `LastError`); reads answer `BadHistoryOperationUnsupported` per node; no host crash |
|
| `Backend=SqlServer`, `AutoCreateSchema=false`, schema missing/behind | every read/write fails with a clear error naming the missing schema version (health `LastError`); reads answer `BadHistoryOperationUnsupported` per node; no host crash |
|
||||||
|
|
||||||
A historized node with no historian configured never returns an error status — it returns
|
A historized node with no historian configured never returns an error status — it returns
|
||||||
@@ -823,6 +824,10 @@ data.
|
|||||||
it could only ever be a Part 11 *interpolated* bound. The start bound on the Gateway is the real prior sample, found by the
|
it could only ever be a Part 11 *interpolated* bound. The start bound on the Gateway is the real prior sample, found by the
|
||||||
#686 walk-back.
|
#686 walk-back.
|
||||||
|
|
||||||
|
**Warm crowding ceiling (#695).** A raw crowding re-read spends the remaining room below a learned gateway ceiling before
|
||||||
|
refusing; doubling the look-back allowance does not skip that final usable budget. Reverse paging retries a crowded segment
|
||||||
|
with a smaller in-window budget derived from the learned ceiling and the observed pre-window rows.
|
||||||
|
|
||||||
### Gateway AtTime is evaluated from raw reads (#693)
|
### Gateway AtTime is evaluated from raw reads (#693)
|
||||||
|
|
||||||
> ⚠️ **Client-visible (#693).** Gateway AtTime used to pass AVEVA's own interpolated answer (the gateway's `ReadAtTime`) through,
|
> ⚠️ **Client-visible (#693).** Gateway AtTime used to pass AVEVA's own interpolated answer (the gateway's `ReadAtTime`) through,
|
||||||
@@ -880,9 +885,14 @@ data.
|
|||||||
here — the SQL backends store DateTime as its own kind and return a `DateTime`. Under an integer hint a NaN, ±∞ or
|
here — the SQL backends store DateTime as its own kind and return a `DateTime`. Under an integer hint a NaN, ±∞ or
|
||||||
out-of-Int64-range stored value is no integer: it is treated as unusable (as `BadNoData`), so a neighbour like that is never
|
out-of-Int64-range stored value is no integer: it is treated as unusable (as `BadNoData`), so a neighbour like that is never
|
||||||
interpolated or saturated into a Good value.
|
interpolated or saturated into a Good value.
|
||||||
- **What differs between backends (by design, not changed by #693).** Only the **Gateway** converts an AtTime result to the node's
|
- **Client value types (#698, all backends).** Before returning Raw, Modified or AtTime history, the node manager converts non-null
|
||||||
integer type (`Int32` ⇒ `int`): the SQL backends store and return integer nodes as `Int64` (and Boolean as Boolean). Gateway
|
scalar values to the node's declared DataType (including Boolean, integer widths and Float). Arrays retain their stored values.
|
||||||
**Raw** reads still return a Boolean node's samples as the stored Double 0/1 (the hint is passed only on AtTime).
|
Conversion failures produce a null value with `BadTypeMismatch` for that sample, preserving timestamps and modification metadata.
|
||||||
|
DateTime requires an actual DateTime value: the gateway's numeric DateTime encoding is unknown and cannot be inferred.
|
||||||
|
Processed results retain each aggregate's result type, so Count and averages can remain Double on a Boolean or integer node.
|
||||||
|
- **Future timestamps (#696).** A bounded raw probe checks for stored samples between now and the requested future time. An empty
|
||||||
|
future interval starts the prior-sample search near now; a full probe seeds the existing walk-back with a stored future sample.
|
||||||
|
The result still carries the requested timestamp and the evaluator's Uncertain status when no later sample exists.
|
||||||
- ⚠️ **Shared-evaluator fix (all backends that use it):** `AtTimeEvaluator` returned every interpolated `Int64` boxed as a
|
- ⚠️ **Shared-evaluator fix (all backends that use it):** `AtTimeEvaluator` returned every interpolated `Int64` boxed as a
|
||||||
**Double** (a C# `cond ? (long)x : double` conditional widens the long); its "keeping the Int64 type" tests passed because
|
**Double** (a C# `cond ? (long)x : double` conditional widens the long); its "keeping the Int64 type" tests passed because
|
||||||
Shouldly's `ShouldBe` compares numbers across types. SqlServer / QuestDB interpolated `Int64` AtTime values are now `Int64`,
|
Shouldly's `ShouldBe` compares numbers across types. SqlServer / QuestDB interpolated `Int64` AtTime values are now `Int64`,
|
||||||
@@ -1764,7 +1774,7 @@ this family and **are** capped (they used to be pushed down). Raise the cap, or
|
|||||||
### Health
|
### Health
|
||||||
|
|
||||||
The read client reports `GetHealthSnapshot()` (query/failure counters, consecutive failures, `LastError` — e.g. a missing schema under
|
The read client reports `GetHealthSnapshot()` (query/failure counters, consecutive failures, `LastError` — e.g. a missing schema under
|
||||||
`AutoCreateSchema=false`). A SQL error on a read surfaces as `BadHistoryOperationUnsupported` for **that node only**; a cancelled read
|
`AutoCreateSchema=false`). A SQL connection outage on a read surfaces as `BadNoCommunication`; other SQL errors retain `BadHistoryOperationUnsupported` for **that node only**; a cancelled read
|
||||||
(deadline, client abort) is not counted as a backend fault.
|
(deadline, client abort) is not counted as a backend fault.
|
||||||
|
|
||||||
### Retention (`RetentionDays`)
|
### Retention (`RetentionDays`)
|
||||||
|
|||||||
@@ -0,0 +1,28 @@
|
|||||||
|
# Historian follow-up results — 2026-10-13
|
||||||
|
|
||||||
|
Scope: approved bounded fixes for #695–#698, plus live measurement of #653. Base: `66a2b0c3`.
|
||||||
|
|
||||||
|
- #695: spend the remaining learned-ceiling room on crowding reads; reduce crowded reverse-read windows using the observed pre-window count. Regression cases cover ceilings 1,709 and 1,369, and reverse paging through a crowded millisecond.
|
||||||
|
- #696 item 1: probe the future interval before starting the prior-sample search near now. Preserve stored future samples, requested timestamps and Uncertain status. Dense future probes still use the existing bounded walk. Items 2 and 3 remain open (stopped-tag strict-ceiling cost and aggregate ceiling re-probes).
|
||||||
|
- #697: actual Gateway, SQL Server and QuestDB transport failures become `BadNoCommunication` on every HistoryRead arm. Cancellation remains `BadTimeout`; unsupported operations retain `BadHistoryOperationUnsupported`. Tests include real refused database connections, repeated reads and nested gRPC failures.
|
||||||
|
- #698: normalize Raw, Modified and AtTime scalar values to the node DataType. Conversion failure gives per-sample null/`BadTypeMismatch`; arrays, audit metadata and timestamps survive. Processed results retain aggregate-specific types.
|
||||||
|
- #694 item 3 remains blocked on historiangw#27. No upstream code was changed.
|
||||||
|
|
||||||
|
## Live #653 measurement
|
||||||
|
|
||||||
|
Five sequential calls per operation and tag, using client 0.3.0 against the running local gateway (`localhost:5221`) with its existing remote AVEVA connection. Commissioned a two-hour `historian:read` key, then revoked it after measurement. No historian writes or provisioning performed.
|
||||||
|
|
||||||
|
| Operation | SysTimeSec, elapsed ms | Nonexistent tag, elapsed ms |
|
||||||
|
|---|---|---|
|
||||||
|
| Metadata | 299, 85, 81, 82, 48 (success) | 89, 53, 41, 77, 76 (gateway exception) |
|
||||||
|
| Raw | 143, 99, 145, 133, 159 (success) | 30260, 30634, 30675, 30754, 30628 (gateway exception) |
|
||||||
|
| Aggregate | 271, 275, 309, 249, 304 (success) | 30564, 30605, 30583, 30575, 30806 (gateway exception) |
|
||||||
|
| AtTime | 140, 88, 89, 127, 92 (success) | 61074, 61330, 61601, 61249, 61271 (gateway exception) |
|
||||||
|
|
||||||
|
Nonexistent tag: `OtOpcUa.Issue653.Nonexistent.20261013`. The delay is still present. Recommend a metadata pre-check with a short negative cache, caching only explicit missing-tag outcomes and never authentication, communication or arbitrary errors. This is follow-up implementation; #653 remains open. Direct gateway AtTime above differs from OtOpcUa's raw-based AtTime evaluator.
|
||||||
|
|
||||||
|
## Consumer and validation notes
|
||||||
|
|
||||||
|
ScadaBridge source and tests contain no HistoryRead consumer; no consumer change required. Independent review found a future-stored-sample edge case, which was fixed and covered before final verification. No deployment or live end-to-end OPC UA verification of these new fixes is claimed; the live run measures #653 through the gateway client.
|
||||||
|
|
||||||
|
Full solution: 14,581 passed, 609 skipped, 0 failed across 62 test suites. Command: `dotnet test ZB.MOM.WW.OtOpcUa.slnx --verbosity quiet`, exit 0. Skips include environment-gated integration suites.
|
||||||
@@ -227,7 +227,27 @@ internal static class GatewayAtTimeReader
|
|||||||
{
|
{
|
||||||
if (!beforeKnown)
|
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;
|
beforeKnown = true;
|
||||||
}
|
}
|
||||||
prev = before;
|
prev = before;
|
||||||
|
|||||||
+24
-1
@@ -152,6 +152,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordReadFailure(ex);
|
RecordReadFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -239,7 +240,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
|||||||
for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++)
|
for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++)
|
||||||
{
|
{
|
||||||
var budget = (int)Math.Min(int.MaxValue, (long)cap + room + RawAggregateReader.PreWindowAllowance);
|
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);
|
pass = await PassAsync(budget).ConfigureAwait(false);
|
||||||
if (pass.Result.Complete) return pass.Result.Samples;
|
if (pass.Result.Complete) return pass.Result.Samples;
|
||||||
preWindow = Math.Max(preWindow, pass.Result.PreWindowRows);
|
preWindow = Math.Max(preWindow, pass.Result.PreWindowRows);
|
||||||
@@ -365,6 +374,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordReadFailure(ex);
|
RecordReadFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -408,6 +418,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordReadFailure(ex);
|
RecordReadFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -466,6 +477,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordReadFailure(ex);
|
RecordReadFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -486,6 +498,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordReadFailure(ex);
|
RecordReadFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -581,6 +594,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordReadFailure(ex);
|
RecordReadFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("HistorianGateway read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -655,6 +669,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
|||||||
return result;
|
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>
|
/// <summary>
|
||||||
/// Records a failed read: bumps the health counters and logs a generic, redaction-safe
|
/// 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.
|
/// 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)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -232,6 +233,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -316,6 +318,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -443,6 +446,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -538,6 +542,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -701,6 +706,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
|||||||
|
|
||||||
private void RecordSuccess() => _health.RecordSuccess();
|
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)
|
private void RecordFailure(Exception ex)
|
||||||
{
|
{
|
||||||
// A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).
|
// 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)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -229,6 +230,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
catch (Exception ex) when (ex is not NotSupportedException)
|
catch (Exception ex) when (ex is not NotSupportedException)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -306,6 +308,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -401,6 +404,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -512,6 +516,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
RecordFailure(ex);
|
RecordFailure(ex);
|
||||||
|
if (IsTransportFailure(ex)) throw new System.IO.IOException("Historian read transport is unavailable.", ex);
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -635,6 +640,9 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
|||||||
|
|
||||||
private void RecordSuccess() => _health.RecordSuccess();
|
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)
|
private void RecordFailure(Exception ex)
|
||||||
{
|
{
|
||||||
// A cancelled read (deadline / client abort) is not a backend fault (the counters ignore it too).
|
// 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);
|
RunBounded(work, HistoryReadBatchParallelism);
|
||||||
});
|
});
|
||||||
|
NormalizeHistoryValues(nodesToProcess, results);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
@@ -5225,6 +5226,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
|||||||
|
|
||||||
RunBounded(work, HistoryReadBatchParallelism);
|
RunBounded(work, HistoryReadBatchParallelism);
|
||||||
});
|
});
|
||||||
|
NormalizeHistoryValues(nodesToProcess, results);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <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
|
// One node's backend failure must not throw out of the batch — surface Bad for THIS node
|
||||||
// only. Logged via the injected ILogger.
|
// only. Logged via the injected ILogger.
|
||||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryReadEvents failed for node {NodeId}", handle.NodeId);
|
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryReadEvents failed for node {NodeId}", handle.NodeId);
|
||||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
errors[handle.Index] = HistoryReadFailureStatus(ex);
|
||||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
|
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
|
// Bad status for THIS node only. Logged through the injected ILogger
|
||||||
// rather than swallowed silently.
|
// rather than swallowed silently.
|
||||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead failed for node {NodeId}", handle.NodeId);
|
_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.
|
// 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);
|
_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)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead (modified) failed for node {NodeId}", handle.NodeId);
|
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead (modified) failed for node {NodeId}", handle.NodeId);
|
||||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
errors[handle.Index] = HistoryReadFailureStatus(ex);
|
||||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
|
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
|
/// <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>
|
/// 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>
|
/// <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
|
var asked = limit > 0 && cap > limit ? limit : cap; // cap 0 (the whole window) is never cut
|
||||||
IReadOnlyList<DataValueSnapshot> samples;
|
IReadOnlyList<DataValueSnapshot> samples;
|
||||||
try
|
for (var attempt = 0; ; attempt++)
|
||||||
{
|
{
|
||||||
samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
|
try
|
||||||
}
|
{
|
||||||
catch (HistoryReadLimitExceededException ex) when (asked > 0 && ex.PreWindowRows > 0 && asked - ex.PreWindowRows >= 1)
|
samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
|
||||||
{
|
break;
|
||||||
// #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.
|
catch (HistoryReadLimitExceededException ex) when (asked > 0 && ex.PreWindowRows > 0 && attempt < 2)
|
||||||
asked -= ex.PreWindowRows;
|
{
|
||||||
samples = (await source.ReadRawAsync(tagname, start, end, (uint)asked, token).ConfigureAwait(false)).Samples;
|
// #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);
|
return new ForwardSegmentRead<DataValueSnapshot>(samples, asked > 0 && samples.Count >= asked);
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -128,6 +128,35 @@ public sealed class GatewayAtTimeTests
|
|||||||
return r.Samples;
|
return r.Samples;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task A_future_time_on_a_live_tag_finds_the_prior_sample_without_walking_the_empty_future()
|
||||||
|
{
|
||||||
|
var store = new AvevaStore(Enumerable.Range(0, 7201).Select(i => (Now.AddSeconds(i - 7200), (double)i, false)));
|
||||||
|
var fake = store.Client();
|
||||||
|
var result = (await AtTime(fake, Now.AddHours(1))).Single();
|
||||||
|
result.Value.ShouldBe(7200d);
|
||||||
|
result.StatusCode.ShouldBe(UncertainInterpolated);
|
||||||
|
fake.ReadRawCallCount.ShouldBeLessThanOrEqualTo(4);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task A_future_time_still_uses_a_stored_future_prior_sample()
|
||||||
|
{
|
||||||
|
var fake = new AvevaStore([(Now.AddSeconds(-1), 1d, false), (Now.AddMinutes(30), 2d, false)]).Client();
|
||||||
|
var result = (await AtTime(fake, Now.AddHours(1))).Single();
|
||||||
|
result.Value.ShouldBe(2d);
|
||||||
|
result.StatusCode.ShouldBe(UncertainInterpolated);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task A_full_future_probe_walks_to_the_latest_stored_future_sample()
|
||||||
|
{
|
||||||
|
var fake = new AvevaStore(Enumerable.Range(0, 7201).Select(i => (Now.AddSeconds(i), (double)i, false))).Client();
|
||||||
|
var result = (await AtTime(fake, Now.AddHours(3))).Single();
|
||||||
|
result.Value.ShouldBe(7200d);
|
||||||
|
result.StatusCode.ShouldBe(UncertainInterpolated);
|
||||||
|
}
|
||||||
|
|
||||||
// ---- the four live findings ------------------------------------------------------------------------------------
|
// ---- the four live findings ------------------------------------------------------------------------------------
|
||||||
|
|
||||||
[Fact] // live: 19:00:03 (a stored NULL) ⇒ 107 Good
|
[Fact] // live: 19:00:03 (a stored NULL) ⇒ 107 Good
|
||||||
|
|||||||
+48
@@ -420,6 +420,54 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
|||||||
finally { await host.DisposeAsync(); }
|
finally { await host.DisposeAsync(); }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData(1709, 1366)] // #695 reverse-walk trace
|
||||||
|
[InlineData(1369, 1292)] // #695 bounds trace
|
||||||
|
public async Task A_warm_read_uses_the_last_room_below_the_ceiling_for_pre_window_ties(int ceiling, int cluster)
|
||||||
|
{
|
||||||
|
var start = At(1000);
|
||||||
|
var stored = Enumerable.Range(0, cluster).Select(i => (start.AddMilliseconds(-0.5), (double)i))
|
||||||
|
.Append((start, 9000d)).Append((start.AddSeconds(1), 9001d)).ToList();
|
||||||
|
var fake = Gateway(stored, ceiling);
|
||||||
|
using var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||||
|
await ds.ReadRawAsync("T", At(0), At(1), 5000, Ct); // learn without crowding
|
||||||
|
fake.RawCalls.Clear();
|
||||||
|
var result = await ds.ReadRawAsync("T", start, start.AddSeconds(2), 1, Ct);
|
||||||
|
result.Samples.Select(s => s.Value).ShouldBe([9000d]);
|
||||||
|
fake.RawCalls.ShouldAllBe(c => c.MaxValues <= ceiling);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task A_warm_reverse_segment_can_shrink_again_when_the_pre_window_cluster_grows()
|
||||||
|
{
|
||||||
|
var start = At(1000);
|
||||||
|
var stored = Enumerable.Range(0, 1366).Select(i => (start.AddMilliseconds(-0.5), (double)i))
|
||||||
|
.Concat(Enumerable.Range(0, 500).Select(i => (start.AddMilliseconds(i), 9000d + i))).ToList();
|
||||||
|
var fake = Gateway(stored, 1709);
|
||||||
|
using var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||||
|
await ds.ReadRawAsync("T", At(0), At(1), 5000, Ct);
|
||||||
|
var page = await ReverseRawReader.ReadAsync(ds, "T", start.AddMilliseconds(499), start.AddTicks(-1),
|
||||||
|
249, 65536, At(2000), new ReverseWalkHint(499 * TimeSpan.TicksPerMillisecond), Ct);
|
||||||
|
page.Samples.Select(s => (double)s.Value!).ShouldBe(Enumerable.Range(9251, 249).Reverse().Select(i => (double)i));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData(false)]
|
||||||
|
[InlineData(true)]
|
||||||
|
public async Task Gateway_transport_faults_surface_as_BadNoCommunication(bool deadline)
|
||||||
|
{
|
||||||
|
var fake = new FakeHistorianGatewayClient { ThrowOnRead = deadline
|
||||||
|
? new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException("deadline", new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.DeadlineExceeded, "deadline")))
|
||||||
|
: new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayUnavailableException("unavailable") };
|
||||||
|
var (host, nm, nodeId) = await BootAsync(fake);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var r = ReadRaw(nm, nodeId, At(0), At(10), 10, null);
|
||||||
|
r.Error.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
|
||||||
|
}
|
||||||
|
finally { await host.DisposeAsync(); }
|
||||||
|
}
|
||||||
|
|
||||||
// ---- helpers ---------------------------------------------------------------------------------------------------------
|
// ---- helpers ---------------------------------------------------------------------------------------------------------
|
||||||
|
|
||||||
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootAsync(
|
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootAsync(
|
||||||
|
|||||||
+27
@@ -11,6 +11,33 @@ public sealed class QuestDbHistorianProviderTests
|
|||||||
ConnectionString = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading",
|
ConnectionString = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading",
|
||||||
});
|
});
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData("Raw")]
|
||||||
|
[InlineData("AtTime")]
|
||||||
|
[InlineData("Processed")]
|
||||||
|
[InlineData("Events")]
|
||||||
|
[InlineData("Modified")]
|
||||||
|
public async Task A_connection_failure_is_exposed_as_a_transport_failure(string kind)
|
||||||
|
{
|
||||||
|
using var provider = new QuestDbHistorianProvider(new QuestDbHistorianOptions {
|
||||||
|
ConnectionString = "Host=127.0.0.1;Port=1;Username=admin;Password=test;Database=qdb;Timeout=1;Pooling=False;ServerCompatibilityMode=NoTypeLoading", AutoCreateSchema = false
|
||||||
|
});
|
||||||
|
using var source = provider.CreateDataSource(new NullSp());
|
||||||
|
var t = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||||
|
Func<Task> read = kind switch
|
||||||
|
{
|
||||||
|
"Raw" => async () => await source.ReadRawAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken),
|
||||||
|
"AtTime" => async () => await source.ReadAtTimeAsync("T", [t], TestContext.Current.CancellationToken),
|
||||||
|
"Processed" => async () => await source.ReadProcessedAsync("T", t, t.AddHours(1), TimeSpan.FromSeconds(10), HistoryAggregateType.Average, TestContext.Current.CancellationToken),
|
||||||
|
"Events" => async () => await source.ReadEventsAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken),
|
||||||
|
_ => async () => await source.ReadModifiedAsync("T", t, t.AddHours(1), 10, null, TestContext.Current.CancellationToken)
|
||||||
|
};
|
||||||
|
var error = await Should.ThrowAsync<System.IO.IOException>(read);
|
||||||
|
error.InnerException.ShouldNotBeNull();
|
||||||
|
// A cached bootstrap outage must remain retryable on subsequent reads as well.
|
||||||
|
(await Should.ThrowAsync<System.IO.IOException>(read)).InnerException.ShouldNotBeNull();
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void Name_and_all_six_kinds()
|
public void Name_and_all_six_kinds()
|
||||||
{
|
{
|
||||||
|
|||||||
+27
@@ -11,6 +11,33 @@ public sealed class SqlServerHistorianProviderTests
|
|||||||
ConnectionString = "Server=localhost;User Id=sa;Password=x;TrustServerCertificate=True",
|
ConnectionString = "Server=localhost;User Id=sa;Password=x;TrustServerCertificate=True",
|
||||||
});
|
});
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData("Raw")]
|
||||||
|
[InlineData("AtTime")]
|
||||||
|
[InlineData("Processed")]
|
||||||
|
[InlineData("Events")]
|
||||||
|
[InlineData("Modified")]
|
||||||
|
public async Task A_connection_failure_is_exposed_as_a_transport_failure(string kind)
|
||||||
|
{
|
||||||
|
var provider = new SqlServerHistorianProvider(new SqlServerHistorianOptions {
|
||||||
|
ConnectionString = "Server=127.0.0.1,1;User Id=sa;Password=test;TrustServerCertificate=True;Connect Timeout=1;ConnectRetryCount=0;Pooling=False", AutoCreateSchema = false
|
||||||
|
});
|
||||||
|
using var source = provider.CreateDataSource(new NullSp());
|
||||||
|
var t = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||||
|
Func<Task> read = kind switch
|
||||||
|
{
|
||||||
|
"Raw" => async () => await source.ReadRawAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken),
|
||||||
|
"AtTime" => async () => await source.ReadAtTimeAsync("T", [t], TestContext.Current.CancellationToken),
|
||||||
|
"Processed" => async () => await source.ReadProcessedAsync("T", t, t.AddHours(1), TimeSpan.FromSeconds(10), HistoryAggregateType.Average, TestContext.Current.CancellationToken),
|
||||||
|
"Events" => async () => await source.ReadEventsAsync("T", t, t.AddHours(1), 10, TestContext.Current.CancellationToken),
|
||||||
|
_ => async () => await source.ReadModifiedAsync("T", t, t.AddHours(1), 10, null, TestContext.Current.CancellationToken)
|
||||||
|
};
|
||||||
|
var error = await Should.ThrowAsync<System.IO.IOException>(read);
|
||||||
|
error.InnerException.ShouldNotBeNull();
|
||||||
|
// A cached bootstrap outage must remain retryable on subsequent reads as well.
|
||||||
|
(await Should.ThrowAsync<System.IO.IOException>(read)).InnerException.ShouldNotBeNull();
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void Name_and_all_six_kinds()
|
public void Name_and_all_six_kinds()
|
||||||
{
|
{
|
||||||
|
|||||||
+9
-5
@@ -295,12 +295,14 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
|
|||||||
/// HistoryRead call. The fake source MUST be invoked (proving we reached the bridge) and threw;
|
/// HistoryRead call. The fake source MUST be invoked (proving we reached the bridge) and threw;
|
||||||
/// the production catch now sets BOTH errors and results explicitly, so both are asserted here
|
/// the production catch now sets BOTH errors and results explicitly, so both are asserted here
|
||||||
/// rather than relying on the SDK base pre-seeding results[i].</summary>
|
/// rather than relying on the SDK base pre-seeding results[i].</summary>
|
||||||
[Fact]
|
[Theory]
|
||||||
public async Task Events_backend_throw_yields_bad_status_and_does_not_escape()
|
[InlineData(false)]
|
||||||
|
[InlineData(true)]
|
||||||
|
public async Task Events_backend_throw_yields_bad_status_and_does_not_escape(bool communication)
|
||||||
{
|
{
|
||||||
var (host, server) = await BootAsync();
|
var (host, server) = await BootAsync();
|
||||||
var nm = server.NodeManager!;
|
var nm = server.NodeManager!;
|
||||||
var fake = new RecordingHistorianDataSource { ThrowOnRead = true };
|
var fake = new RecordingHistorianDataSource { ThrowOnRead = !communication, ReadException = communication ? new System.IO.IOException("lost connection") : null };
|
||||||
nm.HistorianDataSource = fake;
|
nm.HistorianDataSource = fake;
|
||||||
|
|
||||||
const string equipmentId = "eq-boom";
|
const string equipmentId = "eq-boom";
|
||||||
@@ -321,10 +323,10 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
|
|||||||
|
|
||||||
// Authoritative per-node signal: errors[0] must be Bad.
|
// Authoritative per-node signal: errors[0] must be Bad.
|
||||||
ServiceResult.IsBad(errors[0]).ShouldBeTrue();
|
ServiceResult.IsBad(errors[0]).ShouldBeTrue();
|
||||||
errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
errors[0].StatusCode.Code.ShouldBe(communication ? StatusCodes.BadNoCommunication : StatusCodes.BadHistoryOperationUnsupported);
|
||||||
// The production catch now sets results[0] explicitly — assert it directly (not relying on base seeding).
|
// The production catch now sets results[0] explicitly — assert it directly (not relying on base seeding).
|
||||||
results[0].ShouldNotBeNull();
|
results[0].ShouldNotBeNull();
|
||||||
results[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
results[0].StatusCode.Code.ShouldBe(communication ? StatusCodes.BadNoCommunication : StatusCodes.BadHistoryOperationUnsupported);
|
||||||
// The source WAS entered (proving the override reached the bridge and the throw was swallowed).
|
// The source WAS entered (proving the override reached the bridge and the throw was swallowed).
|
||||||
fake.EventsEntered.ShouldBeTrue();
|
fake.EventsEntered.ShouldBeTrue();
|
||||||
// ThrowOnRead fires before EventsCalled/LastSourceName are set — that's the throw path.
|
// ThrowOnRead fires before EventsCalled/LastSourceName are set — that's the throw path.
|
||||||
@@ -380,6 +382,7 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
|
|||||||
private sealed class RecordingHistorianDataSource : IHistorianDataSource
|
private sealed class RecordingHistorianDataSource : IHistorianDataSource
|
||||||
{
|
{
|
||||||
public bool ThrowOnRead { get; init; }
|
public bool ThrowOnRead { get; init; }
|
||||||
|
public Exception? ReadException { get; init; }
|
||||||
public HistoricalEventsResult EventsResult { get; set; } =
|
public HistoricalEventsResult EventsResult { get; set; } =
|
||||||
new(Array.Empty<HistoricalEvent>(), null);
|
new(Array.Empty<HistoricalEvent>(), null);
|
||||||
|
|
||||||
@@ -411,6 +414,7 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
|
|||||||
CancellationToken cancellationToken)
|
CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
EventsEntered = true;
|
EventsEntered = true;
|
||||||
|
if (ReadException is not null) throw ReadException;
|
||||||
if (ThrowOnRead) throw new InvalidOperationException("backend boom for events");
|
if (ThrowOnRead) throw new InvalidOperationException("backend boom for events");
|
||||||
EventsCalled = true;
|
EventsCalled = true;
|
||||||
LastSourceName = sourceName;
|
LastSourceName = sourceName;
|
||||||
|
|||||||
+30
-2
@@ -25,6 +25,34 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable
|
|||||||
private static readonly DateTime Epoch = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
private static readonly DateTime Epoch = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||||
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-modified-{Guid.NewGuid():N}");
|
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-modified-{Guid.NewGuid():N}");
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Modified_values_use_the_scalar_node_type_and_preserve_the_audit_information()
|
||||||
|
{
|
||||||
|
var (host, nm, node, _) = await SetupAsync([Mod(1, 1, 2d, HistoryModificationType.Replace, "alice", Epoch)], "Float");
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var r = Read(nm, node, Epoch, Epoch.AddHours(1), 10, null);
|
||||||
|
r.Data!.DataValues.Single().Value.ShouldBeOfType<float>().ShouldBe(2f);
|
||||||
|
r.Data.ModificationInfos.Single().UserName.ShouldBe("alice");
|
||||||
|
r.Data.ModificationInfos.Single().UpdateType.ShouldBe(HistoryUpdateType.Replace);
|
||||||
|
}
|
||||||
|
finally { await host.DisposeAsync(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task A_modified_read_of_an_unreachable_backend_answers_BadNoCommunication()
|
||||||
|
{
|
||||||
|
var (host, nm, node, src) = await SetupAsync([]);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
src.Throw = new System.IO.IOException("connection lost");
|
||||||
|
var r = Read(nm, node, Epoch, Epoch.AddHours(1), 10, null);
|
||||||
|
r.Error.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
|
||||||
|
r.Result.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
|
||||||
|
}
|
||||||
|
finally { await host.DisposeAsync(); }
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task A_modified_read_dispatches_to_the_trail_and_projects_modification_infos()
|
public async Task A_modified_read_dispatches_to_the_trail_and_projects_modification_infos()
|
||||||
{
|
{
|
||||||
@@ -289,7 +317,7 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable
|
|||||||
}
|
}
|
||||||
|
|
||||||
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId Node, SqlLikeTrail Src)> SetupAsync(
|
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId Node, SqlLikeTrail Src)> SetupAsync(
|
||||||
List<HistoryModifiedValue> trail)
|
List<HistoryModifiedValue> trail, string dataType = "Double")
|
||||||
{
|
{
|
||||||
var started = await OpcUaHostStart.StartAsync(
|
var started = await OpcUaHostStart.StartAsync(
|
||||||
port => new OpcUaApplicationHost(
|
port => new OpcUaApplicationHost(
|
||||||
@@ -307,7 +335,7 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable
|
|||||||
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
|
||||||
var src = new SqlLikeTrail(trail);
|
var src = new SqlLikeTrail(trail);
|
||||||
nm.HistorianDataSource = src;
|
nm.HistorianDataSource = src;
|
||||||
nm.EnsureVariable("eq-1/mod", null, "Mod", "Double", false, historianTagname: Tag, realm: AddressSpaceRealm.Uns);
|
nm.EnsureVariable("eq-1/mod", null, "Mod", dataType, false, historianTagname: Tag, realm: AddressSpaceRealm.Uns);
|
||||||
return (started.Host, nm, nm.TryGetVariable("eq-1/mod")!.NodeId, src);
|
return (started.Host, nm, nm.TryGetVariable("eq-1/mod")!.NodeId, src);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -786,6 +786,132 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
|||||||
return (results, errors);
|
return (results, errors);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData("Raw")]
|
||||||
|
[InlineData("AtTime")]
|
||||||
|
[InlineData("Processed")]
|
||||||
|
public async Task An_unreachable_backend_answers_a_communication_status(string kind)
|
||||||
|
{
|
||||||
|
var (host, server) = await BootAsync();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var nm = server.NodeManager!;
|
||||||
|
nm.HistorianDataSource = new RecordingHistorianDataSource { ReadException = new System.IO.IOException("connection lost") };
|
||||||
|
nm.EnsureVariable("eq/comm", null, "Comm", "Double", false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
|
||||||
|
var t = DateTime.UtcNow.AddMinutes(-1);
|
||||||
|
HistoryReadDetails details = kind switch
|
||||||
|
{
|
||||||
|
"Raw" => new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false, StartTime = t, EndTime = t.AddSeconds(10) },
|
||||||
|
"AtTime" => new ReadAtTimeDetails { ReqTimes = new DateTimeCollection { t } },
|
||||||
|
_ => new ReadProcessedDetails { StartTime = t, EndTime = t.AddSeconds(10), ProcessingInterval = 1000,
|
||||||
|
AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Average } }
|
||||||
|
};
|
||||||
|
var (results, errors) = InvokeHistoryRead(server, nm, details, nm.TryGetVariable("eq/comm")!.NodeId);
|
||||||
|
errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
|
||||||
|
results[0].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
|
||||||
|
}
|
||||||
|
finally { await host.DisposeAsync(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData("Int32", "Raw")]
|
||||||
|
[InlineData("Boolean", "Raw")]
|
||||||
|
[InlineData("Float", "Raw")]
|
||||||
|
[InlineData("Int32", "AtTime")]
|
||||||
|
[InlineData("Boolean", "AtTime")]
|
||||||
|
[InlineData("Float", "AtTime")]
|
||||||
|
public async Task History_values_use_the_nodes_scalar_type(string dataType, string kind)
|
||||||
|
{
|
||||||
|
var (host, server) = await BootAsync();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var nm = server.NodeManager!;
|
||||||
|
var t = DateTime.UtcNow.AddMinutes(-1);
|
||||||
|
var value = new HistorianRead([new DataValueSnapshot(1d, StatusCodes.Good, t, t)], null);
|
||||||
|
nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = value, AtTimeResult = value };
|
||||||
|
nm.EnsureVariable("eq/typed", null, "Typed", dataType, false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
|
||||||
|
HistoryReadDetails details = kind == "Raw"
|
||||||
|
? new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false, StartTime = t, EndTime = t.AddSeconds(10) }
|
||||||
|
: new ReadAtTimeDetails { ReqTimes = new DateTimeCollection { t } };
|
||||||
|
var (results, errors) = InvokeHistoryRead(server, nm, details, nm.TryGetVariable("eq/typed")!.NodeId);
|
||||||
|
errors[0].StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||||
|
var actual = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single();
|
||||||
|
if (dataType == "Int32") actual.Value.ShouldBeOfType<int>().ShouldBe(1);
|
||||||
|
if (dataType == "Boolean") actual.Value.ShouldBeOfType<bool>().ShouldBeTrue();
|
||||||
|
if (dataType == "Float") actual.Value.ShouldBeOfType<float>().ShouldBe(1f);
|
||||||
|
actual.SourceTimestamp.ShouldBe(t);
|
||||||
|
actual.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||||
|
}
|
||||||
|
finally { await host.DisposeAsync(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData("Int16", 40000d)]
|
||||||
|
[InlineData("DateTime", 1d)]
|
||||||
|
public async Task An_unrepresentable_historical_value_is_null_with_BadTypeMismatch(string type, double input)
|
||||||
|
{
|
||||||
|
var (host, server) = await BootAsync();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var nm = server.NodeManager!;
|
||||||
|
var t = DateTime.UtcNow.AddMinutes(-1);
|
||||||
|
nm.EnsureVariable("eq/mismatch", null, "Mismatch", type, false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
|
||||||
|
nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = new HistorianRead([new(input, StatusCodes.Good, t, t)], null) };
|
||||||
|
var (results, _) = InvokeHistoryRead(server, nm, new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false,
|
||||||
|
StartTime = t, EndTime = t.AddSeconds(10) }, nm.TryGetVariable("eq/mismatch")!.NodeId);
|
||||||
|
var value = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single();
|
||||||
|
value.Value.ShouldBeNull();
|
||||||
|
value.StatusCode.Code.ShouldBe(StatusCodes.BadTypeMismatch);
|
||||||
|
value.SourceTimestamp.ShouldBe(t);
|
||||||
|
value.ServerTimestamp.ShouldBe(t);
|
||||||
|
}
|
||||||
|
finally { await host.DisposeAsync(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Array_and_null_history_values_preserve_quality_and_timestamps()
|
||||||
|
{
|
||||||
|
var (host, server) = await BootAsync();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var nm = server.NodeManager!;
|
||||||
|
var t = DateTime.UtcNow.AddMinutes(-1);
|
||||||
|
var serverTime = t.AddSeconds(1);
|
||||||
|
nm.EnsureVariable("eq/array", null, "Array", "Int32", false, historianTagname: "T", realm: AddressSpaceRealm.Raw, isArray: true, arrayLength: 2);
|
||||||
|
nm.HistorianDataSource = new RecordingHistorianDataSource { RawResult = new HistorianRead([
|
||||||
|
new(new int[] { 1, 2 }, StatusCodes.Uncertain, t, serverTime),
|
||||||
|
new(null, StatusCodes.BadNoCommunication, t.AddSeconds(1), serverTime)], null) };
|
||||||
|
var (results, _) = InvokeHistoryRead(server, nm, new ReadRawModifiedDetails { IsReadModified = false, ReturnBounds = false,
|
||||||
|
StartTime = t, EndTime = t.AddSeconds(10) }, nm.TryGetVariable("eq/array", AddressSpaceRealm.Raw)!.NodeId);
|
||||||
|
var values = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues;
|
||||||
|
values[0].Value.ShouldBeOfType<int[]>().ShouldBe([1, 2]);
|
||||||
|
values[0].StatusCode.Code.ShouldBe(StatusCodes.Uncertain);
|
||||||
|
values[0].SourceTimestamp.ShouldBe(t);
|
||||||
|
values[0].ServerTimestamp.ShouldBe(serverTime);
|
||||||
|
values[1].Value.ShouldBeNull();
|
||||||
|
values[1].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
|
||||||
|
}
|
||||||
|
finally { await host.DisposeAsync(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Processed_Count_keeps_its_aggregate_type_on_a_Boolean_node()
|
||||||
|
{
|
||||||
|
var (host, server) = await BootAsync();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var nm = server.NodeManager!;
|
||||||
|
var t = DateTime.UtcNow.AddMinutes(-1);
|
||||||
|
nm.EnsureVariable("eq/count", null, "Count", "Boolean", false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
|
||||||
|
nm.HistorianDataSource = new RecordingHistorianDataSource { ProcessedResult = new HistorianRead([new(3d, StatusCodes.Good, t, t)], null) };
|
||||||
|
var (results, _) = InvokeHistoryRead(server, nm, new ReadProcessedDetails { StartTime = t, EndTime = t.AddSeconds(10),
|
||||||
|
ProcessingInterval = 1000, AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Count } }, nm.TryGetVariable("eq/count")!.NodeId);
|
||||||
|
var value = ((HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData)).DataValues.Single();
|
||||||
|
value.Value.ShouldBeOfType<double>().ShouldBe(3d);
|
||||||
|
}
|
||||||
|
finally { await host.DisposeAsync(); }
|
||||||
|
}
|
||||||
|
|
||||||
private static HistorianRead Empty() => new(Array.Empty<DataValueSnapshot>(), null);
|
private static HistorianRead Empty() => new(Array.Empty<DataValueSnapshot>(), null);
|
||||||
|
|
||||||
/// <summary>A recording fake historian source — captures the last call's kind + arguments and returns
|
/// <summary>A recording fake historian source — captures the last call's kind + arguments and returns
|
||||||
@@ -793,6 +919,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
|||||||
private sealed class RecordingHistorianDataSource : IHistorianDataSource
|
private sealed class RecordingHistorianDataSource : IHistorianDataSource
|
||||||
{
|
{
|
||||||
public bool ThrowOnRead { get; init; }
|
public bool ThrowOnRead { get; init; }
|
||||||
|
public Exception? ReadException { get; init; }
|
||||||
public bool AggregateNotSupported { get; init; }
|
public bool AggregateNotSupported { get; init; }
|
||||||
public bool LimitExceeded { get; init; }
|
public bool LimitExceeded { get; init; }
|
||||||
public IReadOnlySet<HistoryAggregateType>? Supported { get; init; } = HistoryAggregateTypes.All;
|
public IReadOnlySet<HistoryAggregateType>? Supported { get; init; } = HistoryAggregateTypes.All;
|
||||||
@@ -814,6 +941,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
|||||||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||||||
CancellationToken cancellationToken)
|
CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
|
if (ReadException is not null) throw ReadException;
|
||||||
if (ThrowOnRead) throw new InvalidOperationException("backend boom");
|
if (ThrowOnRead) throw new InvalidOperationException("backend boom");
|
||||||
LastCall = "Raw";
|
LastCall = "Raw";
|
||||||
LastTagname = fullReference;
|
LastTagname = fullReference;
|
||||||
@@ -827,6 +955,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
|||||||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||||||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
|
if (ReadException is not null) throw ReadException;
|
||||||
if (ThrowOnRead) throw new InvalidOperationException("backend boom");
|
if (ThrowOnRead) throw new InvalidOperationException("backend boom");
|
||||||
if (AggregateNotSupported) throw new HistoryAggregateNotSupportedException(aggregate);
|
if (AggregateNotSupported) throw new HistoryAggregateNotSupportedException(aggregate);
|
||||||
if (LimitExceeded) throw new HistoryReadLimitExceededException(4, "too many raw rows");
|
if (LimitExceeded) throw new HistoryReadLimitExceededException(4, "too many raw rows");
|
||||||
@@ -842,6 +971,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
|||||||
public Task<HistorianRead> ReadAtTimeAsync(
|
public Task<HistorianRead> ReadAtTimeAsync(
|
||||||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
|
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
|
if (ReadException is not null) throw ReadException;
|
||||||
if (ThrowOnRead) throw new InvalidOperationException("backend boom");
|
if (ThrowOnRead) throw new InvalidOperationException("backend boom");
|
||||||
LastCall = "AtTime";
|
LastCall = "AtTime";
|
||||||
LastTagname = fullReference;
|
LastTagname = fullReference;
|
||||||
|
|||||||
Reference in New Issue
Block a user