docs(historian): pluggable historian backends — design + implementation plan
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,332 @@
|
||||
# Pluggable historian backends — design
|
||||
|
||||
**Date:** 2026-10-09 · **Status:** approved (brainstorm 2026-10-09) · **Plan:** `2026-10-09-pluggable-historian-plan.md`
|
||||
|
||||
## 1. Goal and scope
|
||||
|
||||
Today the only historian backend is AVEVA/Wonderware, reached through the HistorianGateway sidecar. This
|
||||
work makes the backend **pluggable**, ships **SQL Server** as the first new backend, and adds the OPC UA
|
||||
**HistoryUpdate** service so clients can insert / replace / update / delete raw history.
|
||||
|
||||
| In scope (this spec) | Later (own spec) | Dropped |
|
||||
|---|---|---|
|
||||
| Provider seam + `ServerHistorian:Backend` selection | QuestDB backend (a new provider only) | SQLite (a per-node file needs pair replication) |
|
||||
| Typed values (not just `double`) end to end through recorder + outbox | | |
|
||||
| Real status codes recorded (fixes a hard-coded Good) | | |
|
||||
| SQL Server backend: schema, writes, reads, aggregates push-down, alarm events, provisioning, updates | | |
|
||||
| HistoryUpdate service + `HistoryUpdate` role / permission + AdminUI checkbox | | |
|
||||
| Client.CLI `historyupdate` command | | |
|
||||
|
||||
Out of scope everywhere: retention/purge, partitioning/compression, Part 11 "modified values"
|
||||
(`IsReadModified`, `IsDeleteModified`), event HistoryUpdate (`UpdateEventDetails`/`DeleteEventDetails`),
|
||||
`UpdateStructureData`, `AuditHistoryUpdateEvent` emission, array historization.
|
||||
|
||||
Hard constraints: no new NuGet packages and no version changes (`Microsoft.Data.SqlClient` is already
|
||||
centrally pinned — referencing it from a new project is allowed, adding a new package is not).
|
||||
Recording stays Primary-only (#537); delivery stays at-least-once with the #538 per-tag quarantine.
|
||||
|
||||
## 2. Architecture — the provider seam (approach A)
|
||||
|
||||
The five historian seams the rest of the code already consumes stay; a provider builds them.
|
||||
|
||||
```csharp
|
||||
// src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianBackendProvider.cs
|
||||
public interface IHistorianBackendProvider
|
||||
{
|
||||
string Name { get; } // "Gateway" | "SqlServer" | (later) "QuestDb"
|
||||
IReadOnlySet<HistorianValueKind> SupportedKinds { get; } // recorder capture filter
|
||||
IHistorianDataSource CreateDataSource(IServiceProvider sp);
|
||||
IHistorianValueWriter CreateValueWriter(IServiceProvider sp);
|
||||
IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp);
|
||||
IHistorianProvisioning CreateProvisioning(IServiceProvider sp);
|
||||
IHistorianDataUpdater? CreateUpdater(IServiceProvider sp); // null ⇒ HistoryUpdate unsupported
|
||||
}
|
||||
```
|
||||
|
||||
It lives in **Runtime** because the seams it returns span `Core.Abstractions` and `Core.AlarmHistorian`.
|
||||
|
||||
- `GatewayHistorianProvider` (Driver.Historian.Gateway) wraps today's `GatewayHistorian.Create…` factories;
|
||||
`CreateUpdater` returns `null`; `SupportedKinds = {Double, Int64, Boolean}`.
|
||||
- `SqlServerHistorianProvider` (new project `Driver.Historian.SqlServer`); `SupportedKinds` = all kinds.
|
||||
- The Host owns a name→provider table (`HistorianBackendRegistry`). `Program.cs`'s four hard-wired
|
||||
`GatewayHistorian.Create…` call sites and the inline `new GatewayHistorianValueWriter(...)` become calls
|
||||
on the selected provider. The existing gates (`AddServerHistorian`, `AddHistorianProvisioning`,
|
||||
`AddAlarmHistorian`, the outbox registration, the recorder spawn) are unchanged — they get their objects
|
||||
from the provider.
|
||||
- The outbox (`FasterLogHistorizationOutbox`) stays in the Gateway project (moving it would add a
|
||||
FASTER package reference elsewhere); the Host registers it regardless of backend. The SqlServer project
|
||||
does **not** reference the Gateway project.
|
||||
- Rejected: one fat `IHistorian` (forces stubs, churns every narrow consumer); per-capability backend
|
||||
selection (YAGNI, doubles config + validation).
|
||||
|
||||
### Configuration
|
||||
|
||||
```jsonc
|
||||
"ServerHistorian": {
|
||||
"Enabled": true,
|
||||
"Backend": "SqlServer", // "Gateway" (default when absent) | "SqlServer"
|
||||
"HistoryReadBatchParallelism": 4, "MaxConcurrentHistoryReadBatches": 16, "HistoryReadDeadline": "00:01:00",
|
||||
"Endpoint": "...", "ApiKey": "...", // gateway keys stay top-level, unchanged (no migration)
|
||||
"SqlServer": {
|
||||
"ConnectionString": "${secret:OT-HISTORIAN-SQL}",
|
||||
"Schema": "hist",
|
||||
"CommandTimeout": "00:00:30",
|
||||
"AutoCreateSchema": true
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- Unknown `Backend` ⇒ host start fails (`OptionsValidationException` listing the valid names).
|
||||
- `ServerHistorianOptionsValidator` requires `Endpoint` only when `Backend == Gateway`.
|
||||
- New `SqlServerHistorianOptions` validation: `ConnectionString` required, `Schema` a valid SQL identifier
|
||||
(`^[A-Za-z_][A-Za-z0-9_]{0,127}$` — it is spliced into DDL/DML), `CommandTimeout > 0`.
|
||||
- Validators remain driver-block-only (admin-only nodes never build historian singletons).
|
||||
- The `Program.cs` alarm-only warning names the selected backend instead of "gateway".
|
||||
- The historian connection string is **not** a ConfigDb string; driver-only nodes may carry it (Phase 4 intact).
|
||||
|
||||
## 3. Typed values
|
||||
|
||||
```csharp
|
||||
// Core.Abstractions/Historian/HistorianValue.cs
|
||||
public enum HistorianValueKind : byte { Empty = 0, Double = 1, Int64 = 2, Boolean = 3, String = 4, DateTime = 5 }
|
||||
|
||||
public readonly record struct HistorianValue(HistorianValueKind Kind, double Number, long Integer, string? Text)
|
||||
{
|
||||
public static HistorianValue Empty { get; } // no value (a Bad/Uncertain sample with a null)
|
||||
public static HistorianValue FromDouble(double v);
|
||||
public static HistorianValue FromInt64(long v);
|
||||
public static HistorianValue FromBoolean(bool v); // Integer = 1/0, Number = 1/0
|
||||
public static HistorianValue FromString(string v);
|
||||
public static HistorianValue FromDateTime(DateTime utc); // Integer = Ticks (UTC)
|
||||
public static bool TryFromObject(object? value, out HistorianValue result); // the recorder's coercion
|
||||
public bool TryGetDouble(out double v); // Double/Int64/Boolean ⇒ true
|
||||
public object? ToObject(); // back to a CLR value for DataValueSnapshot
|
||||
}
|
||||
```
|
||||
|
||||
`TryFromObject` mapping: `double/float/decimal` ⇒ Double; `sbyte/byte/short/ushort/int/uint/long` ⇒ Int64;
|
||||
`ulong` ⇒ Int64 when `≤ long.MaxValue` else Double; `bool` ⇒ Boolean; `string` ⇒ String; `DateTime` ⇒
|
||||
DateTime (converted to UTC); `null` ⇒ Empty; arrays / anything else ⇒ `false` (dropped + metered).
|
||||
|
||||
Changed records:
|
||||
|
||||
```csharp
|
||||
public readonly record struct HistorizationValue(DateTime? TimestampUtc, HistorianValue Value, uint StatusCode);
|
||||
public sealed record HistorizationOutboxEntry(Guid Id, string Tag, HistorianValue Value, uint StatusCode, DateTime TimestampUtc);
|
||||
```
|
||||
|
||||
**Recorder.** Captures `HistorianValue` via `TryFromObject`; records `msg.StatusCode` (the mux already
|
||||
carries it — the recorder hard-codes `GoodQuality = 192` today, so Bad samples were historized as Good).
|
||||
A value whose kind is not in the provider's `SupportedKinds` is dropped at capture and counted under
|
||||
`DroppedUnsupportedKind` (renamed from `DroppedNonNumeric`, same meter semantics for the gateway). `Empty`
|
||||
is in SQL Server's set (a Bad sample with no value is history worth keeping) and not the gateway's.
|
||||
|
||||
**Gateway backend: no wire change in practice.** Samples now carry their real status, mapped to DA quality
|
||||
inside `GatewayHistorianValueWriter` by a new reverse mapper `GatewayQualityMapper.ToDaQuality(uint)` — the
|
||||
exact inverse of the existing `Map(byte)` table, falling back by severity (Good ⇒ 192, Uncertain ⇒ 64,
|
||||
Bad ⇒ 0). The recorder no longer knows about DA quality. But the gateway's `WriteLiveValues` SQL path
|
||||
**discards** `Quality` (Contracts C-002, server-managed), so AVEVA keeps storing every recorded value as Good —
|
||||
a pre-existing limitation of that backend, now documented rather than fixed. The SQL Server backend stores
|
||||
the real status.
|
||||
|
||||
**Outbox format v2.** Entries are written `magic "HV2\0"` + `version (1 byte)` + id(16) + tagLen(4) + tag +
|
||||
kind(1) + kind payload + status(4) + timestamp(8). Payload by kind: Empty none; Double 8; Int64/Boolean/DateTime
|
||||
8; String len(4)+UTF-8. Legacy entries (no magic, `length == 38 + tagLen@16`) are read as Double with
|
||||
status `GatewayQualityMapper.Map(quality)` (DA ⇒ StatusCode; legacy entries are always 192 ⇒ Good). An entry
|
||||
matching neither layout is dropped and counted (`otopcua.historian.outbox.unreadable`), never thrown — a
|
||||
corrupt entry must not jam the drain. A pinned legacy-bytes test guards the upgrade path.
|
||||
|
||||
## 4. SQL Server backend
|
||||
|
||||
### Schema (`{Schema}` default `hist`, database e.g. `OtOpcUaHistorian`)
|
||||
|
||||
```sql
|
||||
CREATE TABLE {s}.SchemaVersion (Version int NOT NULL PRIMARY KEY, AppliedUtc datetime2(7) NOT NULL);
|
||||
|
||||
CREATE TABLE {s}.Tag (
|
||||
TagId int IDENTITY(1,1) NOT NULL PRIMARY KEY,
|
||||
Name nvarchar(400) NOT NULL CONSTRAINT UQ_Tag_Name UNIQUE,
|
||||
DataType tinyint NOT NULL, -- DriverDataType
|
||||
EngineeringUnit nvarchar(64) NULL, Description nvarchar(400) NULL,
|
||||
CreatedUtc datetime2(7) NOT NULL, UpdatedUtc datetime2(7) NOT NULL);
|
||||
|
||||
CREATE TABLE {s}.Sample (
|
||||
TagId int NOT NULL, SourceTimeUtc datetime2(7) NOT NULL,
|
||||
ValueKind tinyint NOT NULL, -- HistorianValueKind
|
||||
NumValue float NULL, IntValue bigint NULL, StrValue nvarchar(max) NULL,
|
||||
StatusCode int NOT NULL, -- uint StatusCode stored bit-for-bit (unchecked cast)
|
||||
ServerTimeUtc datetime2(7) NOT NULL,
|
||||
CONSTRAINT PK_Sample PRIMARY KEY CLUSTERED (TagId, SourceTimeUtc),
|
||||
CONSTRAINT FK_Sample_Tag FOREIGN KEY (TagId) REFERENCES {s}.Tag(TagId));
|
||||
|
||||
CREATE TABLE {s}.Event (
|
||||
EventId nvarchar(64) NOT NULL PRIMARY KEY, -- deterministic SHA-256 hex of the event payload
|
||||
SourceName nvarchar(400) NOT NULL, -- AlarmHistorianEvent.EquipmentPath
|
||||
AlarmId nvarchar(400) NOT NULL, AlarmName nvarchar(400) NOT NULL, AlarmTypeName nvarchar(100) NOT NULL,
|
||||
Severity int NOT NULL, EventKind nvarchar(50) NOT NULL, Message nvarchar(max) NULL,
|
||||
UserName nvarchar(200) NULL, Comment nvarchar(max) NULL,
|
||||
EventTimeUtc datetime2(7) NOT NULL, ReceivedTimeUtc datetime2(7) NOT NULL);
|
||||
CREATE INDEX IX_Event_Source_Time ON {s}.Event (SourceName, EventTimeUtc);
|
||||
CREATE INDEX IX_Event_Time ON {s}.Event (EventTimeUtc);
|
||||
|
||||
CREATE TYPE {s}.SampleRow AS TABLE (
|
||||
SourceTimeUtc datetime2(7) NOT NULL PRIMARY KEY, ValueKind tinyint NOT NULL,
|
||||
NumValue float NULL, IntValue bigint NULL, StrValue nvarchar(max) NULL, StatusCode int NOT NULL);
|
||||
CREATE TYPE {s}.TimeRow AS TABLE (T datetime2(7) NOT NULL PRIMARY KEY);
|
||||
```
|
||||
|
||||
Column mapping per kind: Double ⇒ NumValue; Int64 ⇒ IntValue + NumValue (double, for aggregates);
|
||||
Boolean ⇒ IntValue 0/1 + NumValue 0/1; String ⇒ StrValue; DateTime ⇒ IntValue (ticks); Empty ⇒ all NULL.
|
||||
|
||||
The PK enforces Part 11's one-value-per-timestamp and makes outbox redelivery idempotent: live writes are
|
||||
**insert-if-absent**. Events are insert-if-absent on `EventId` (`AlarmHistorianEvent` carries no id, so
|
||||
the writer computes `SHA-256(AlarmId|EventKind|TimestampUtc.Ticks|User|Comment|Message)`, hex).
|
||||
|
||||
**Bootstrap.** Numbered idempotent scripts (`001_initial.sql`) are embedded resources (and mirrored to
|
||||
`docs/historian/sqlserver/`). `SqlServerSchemaBootstrapper` runs once per process on first use, lazily,
|
||||
under `sp_getapplock @Resource='OtOpcUaHistorianSchema', @LockMode='Exclusive', @LockOwner='Session'`.
|
||||
`AutoCreateSchema=true` creates the schema + applies missing scripts; `false` only checks
|
||||
`SchemaVersion` and, if behind, every historian call fails with a clear error naming the missing version
|
||||
(surfaced in `GetHealthSnapshot().LastError`, never a host crash). Bootstrap failure is retried on the
|
||||
next call (not latched).
|
||||
|
||||
### Writes
|
||||
|
||||
- `IHistorianValueWriter.WriteLiveValuesAsync(tag, values)`: resolve `TagId` via a
|
||||
`ConcurrentDictionary<string,int>` cache, creating the tag row on a miss (`MERGE` on Name; DataType from
|
||||
the first value's kind, provisioning later corrects it). Insert via `{s}.SampleRow` TVP with
|
||||
`INSERT … SELECT … WHERE NOT EXISTS`. Returns `true` on success, `false` on a SQL failure (the recorder's
|
||||
retry/quarantine owns retry). `ServerTimeUtc = SYSUTCDATETIME()`.
|
||||
- `IHistorianProvisioning.EnsureTagsAsync`: `MERGE` upsert of every request (all types — no numeric
|
||||
filter on this backend). Per-request failures are counted, never thrown.
|
||||
- `IAlarmHistorianWriter.WriteBatchAsync`: per event insert-if-absent; outcome `Ack` on insert or
|
||||
already-present, `RetryPlease` on transient SQL errors (connection/timeout/deadlock 1205), `PermanentFail`
|
||||
on data errors (e.g. truncation 2628/8152).
|
||||
|
||||
### Reads
|
||||
|
||||
- **Raw:** `SELECT TOP(@max) … WHERE TagId=@id AND SourceTimeUtc >= @lo AND SourceTimeUtc < @hi ORDER BY
|
||||
SourceTimeUtc ASC` (forward) or `> @lo AND <= @hi … DESC` when `endUtc < startUtc` (Part 11 reverse).
|
||||
Unknown tag ⇒ empty result (node manager reports `GoodNoData`). The node manager's paging is unchanged;
|
||||
tie clusters cannot occur (PK).
|
||||
- **AtTime:** one statement, `{s}.TimeRow` TVP + two `OUTER APPLY`s (previous-or-equal, next). Exact ⇒ stored
|
||||
value + status. Otherwise Double/Int64 linearly interpolated (status `Good` with the historian
|
||||
*Interpolated* info bits), Boolean/String/DateTime stepped from the previous
|
||||
sample (status keeps the previous sample's code + Interpolated bit); no previous sample ⇒ `BadNoData`.
|
||||
If either neighbour is Bad, the result is `UncertainDataSubNormal`.
|
||||
- **Events:** `TOP(@maxEvents)` from `{s}.Event` filtered by `SourceName = @source` (null ⇒ all) and
|
||||
`EventTimeUtc` range, ordered ascending. `HistoricalEvent(EventId, SourceName, EventTimeUtc,
|
||||
ReceivedTimeUtc, Message, (ushort)Severity)`.
|
||||
- **Processed (push-down):** bucket `b = DATEDIFF_BIG(millisecond, @start, SourceTimeUtc) / @intervalMs`,
|
||||
one row per bucket start `@start + b × interval` for every interval in `[start, end)` (empty intervals
|
||||
emitted as `BadNoData`). Usable sample = status severity bits Good (`00`) or Uncertain (`01`). Every result
|
||||
carries the historian *Calculated* info bits. An interval with any Bad
|
||||
sample but ≥1 usable ⇒ `UncertainDataSubNormal`; no usable ⇒ `BadNoData`.
|
||||
|
||||
| Aggregate | SQL over usable samples in the bucket | Value timestamp |
|
||||
|---|---|---|
|
||||
| Average / Minimum / Maximum / Total / Count | `AVG / MIN / MAX / SUM (NumValue)`, `COUNT(*)` | bucket start |
|
||||
| Start / End | first / last by `SourceTimeUtc` (`ROW_NUMBER`) | bucket start |
|
||||
| Delta | End − Start | bucket start |
|
||||
| Range | Max − Min | bucket start |
|
||||
| MinimumActualTime / MaximumActualTime | min / max value, ties ⇒ earliest | the sample's time |
|
||||
| PercentGood / PercentBad | `100 × count(good or bad) / count(*)` over ALL samples | bucket start |
|
||||
|
||||
Count / Start / End / PercentGood / PercentBad accept any kind; the rest on a String/DateTime tag ⇒
|
||||
`BadAggregateInvalidInputs` per interval. Any other `HistoryAggregateType` ⇒ throw
|
||||
`HistoryAggregateNotSupportedException` (⇒ `BadAggregateNotSupported`). These are the simple per-sample
|
||||
definitions, not Part 13's full interval-boundary semantics — documented as such.
|
||||
|
||||
### Updates (`IHistorianDataUpdater`)
|
||||
|
||||
```csharp
|
||||
public enum HistoryUpdateMode { Insert, Replace, Update } // Part 11 PerformUpdateType minus Remove
|
||||
public interface IHistorianDataUpdater
|
||||
{
|
||||
Task<IReadOnlyList<uint>> UpdateDataAsync(string tag, HistoryUpdateMode mode,
|
||||
IReadOnlyList<HistorizationValue> values, CancellationToken ct); // one code per value, in order
|
||||
Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct);
|
||||
Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc, CancellationToken ct);
|
||||
}
|
||||
```
|
||||
|
||||
| Operation | Per-item result |
|
||||
|---|---|
|
||||
| Insert | `GoodEntryInserted`; `BadEntryExists` if a sample exists at that timestamp |
|
||||
| Replace | `GoodEntryReplaced`; `BadNoEntryExists` if none |
|
||||
| Update | `GoodEntryInserted` or `GoodEntryReplaced` |
|
||||
| DeleteRaw | `Good`; `BadNoData` if the range held nothing |
|
||||
| DeleteAtTime | per time: `Good`; `BadNoEntryExists` |
|
||||
|
||||
Each call is one transaction (`MERGE … OUTPUT $action` against the TVP). A value with a null timestamp ⇒
|
||||
`BadInvalidTimestamp` for that item. Update/Insert on an unknown tag creates the tag row (same as live
|
||||
writes); Replace/Delete on an unknown tag ⇒ `BadNoEntryExists` / `BadNoData`.
|
||||
|
||||
**Status codes.** Every status this backend produces is a named `const uint` in an internal
|
||||
`SqlHistorianStatusCodes` class (the house rule: `StatusCodeParityTests` guards named constants by reflection
|
||||
against `Opc.Ua.StatusCodes`, and an inline literal is invisible to it) — register the class with that parity
|
||||
test. Historian info bits follow Part 4 §7.39: `InfoType = DataValue (0x00000400)` OR-ed with
|
||||
`HistorianCalculated (0x1)` / `HistorianInterpolated (0x2)`; the constants are checked against the SDK's
|
||||
`AggregateBits`/`StatusCode` helpers in a unit test rather than trusted from this doc.
|
||||
|
||||
## 5. HistoryUpdate in the server
|
||||
|
||||
- `OtOpcUaNodeManager` overrides the SDK's protected `HistoryUpdateData`, `HistoryDeleteRawModified`,
|
||||
`HistoryDeleteAtTime` (verify exact signatures against the pinned SDK before writing; the read overrides
|
||||
at `OtOpcUaNodeManager.cs` ~4928 are the pattern). The node → historian tagname lookup is the existing
|
||||
`TryGetHistorizedTagname`, so raw and UNS NodeIds hit the same rows. Not historized ⇒
|
||||
`BadHistoryOperationUnsupported`.
|
||||
- New settable `IHistorianDataUpdater? HistorianDataUpdater` on the node manager, assigned by
|
||||
`OtOpcUaServerHostedService` next to `HistorianDataSource` (resolved as optional — `null` when the
|
||||
backend has none). Null ⇒ every HistoryUpdate item `BadHistoryOperationUnsupported`.
|
||||
- AccessLevel: `ComposeAccessLevel(writable, historized)` gains `historyWritable` — the `HistoryWrite` bit
|
||||
is OR-ed in when historized **and** an updater is wired (both raw node and UNS mirrors).
|
||||
- Out-of-scope details (events, structure data, `IsDeleteModified`) ⇒ `BadHistoryOperationUnsupported`.
|
||||
- Updates go straight to the backend (not the outbox), are not Primary-gated (the database is shared),
|
||||
and share the HistoryRead batch limiter + deadline.
|
||||
- **Authorization.**
|
||||
- ACL flag **on**: `DataPlaneAuthorizer.Authorize(identity, OpcUaOperation.HistoryUpdate, nodeId, realm)`
|
||||
(`TriePermissionEvaluator` already maps it to `NodePermissions.HistoryUpdate`).
|
||||
- ACL flag **off**: new role `OpcUaDataPlaneRoles.HistoryUpdate = "HistoryUpdate"`, conferred by
|
||||
`ApiKeyInterimRoleMapper` only from a **Cluster-scoped** grant carrying `NodePermissions.HistoryUpdate`,
|
||||
plus `AuthorizeKeyUsable` (#563 key re-check). Deny ⇒ `BadUserAccessDenied`.
|
||||
- **Load-bearing test:** an ALLOW — flag on, session with no roles, scoped `HistoryUpdate` grant ⇒ the
|
||||
insert lands. And flag off: a `WriteOperate`-only key is DENIED (the role is not reused).
|
||||
- `NodePermissions.HistoryUpdate` is added to the AdminUI ACL editor's permission list (`AclEdit.razor`);
|
||||
it stays out of every composite bundle.
|
||||
|
||||
## 6. Client.CLI
|
||||
|
||||
`historyupdate` command: `-u <url> -n <nodeId> --mode insert|replace|update|delete-raw|delete-at-time`,
|
||||
`--value <v> --time <iso>` (repeatable for insert/replace/update/delete-at-time), `--start/--end` for
|
||||
delete-raw. Prints one line per item with its status code name. Docs in `docs/Client.CLI.md`.
|
||||
|
||||
## 7. Testing
|
||||
|
||||
- **Unit:** `HistorianValue` coercion table; outbox v2 round-trip per kind + pinned legacy bytes +
|
||||
unreadable-entry drop; recorder typed capture, real status passthrough, `SupportedKinds` filter (incl.
|
||||
`Empty`), unchanged #537/#538 behaviour; gateway writer conversion (Good bytes identical to today);
|
||||
provider registry + validators; role mapper; node manager HistoryUpdate (fake updater): allow/deny matrix
|
||||
for flag on/off, result-code passthrough, unsupported cases, AccessLevel bit.
|
||||
- **SQL Server integration** (`Driver.Historian.SqlServer.IntegrationTests`): env-gated on
|
||||
`HISTORIAN_SQL_TEST_CONNECTION` (a master-capable connection string; absent ⇒ `Assert.Skip`), each run in
|
||||
a throwaway database `OtOpcUaHistorian_T_<guid>` dropped on dispose. Default target: the docker-dev
|
||||
`sql` container, `Server=localhost,14330;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True`.
|
||||
Covers bootstrap (idempotent; two concurrent bootstrappers), every kind round-trip, redelivery idempotency,
|
||||
raw forward/reverse/limit, AtTime exact/interpolated/stepped/no-prior, each aggregate against hand-computed
|
||||
values incl. Bad-sample and empty-interval cases, events, every update/delete result code, unknown tags.
|
||||
- **Over-the-wire:** one `OpcUaServer.IntegrationTests` test drives HistoryUpdate Insert then HistoryRead
|
||||
through a real SDK client against a fake updater/data source.
|
||||
- **Live gate (docker-dev):** central pair on `Backend=SqlServer` against database `OtOpcUaHistorian` on
|
||||
the rig `sql` container; a historized rig tag records rows (Primary only — row count advances once per
|
||||
value); Client.CLI `historyread` raw + processed; `historyupdate` insert/replace/delete round-trip;
|
||||
stop the Primary ⇒ the Secondary takes over recording into the same table with no duplicate rows; a
|
||||
Bad sample is stored with its Bad status. Evidence recorded in `docs/plans/2026-10-09-pluggable-historian-live-gate.md`.
|
||||
|
||||
## 8. Docs and index
|
||||
|
||||
`docs/Historian.md` (backend selection, SQL schema, aggregate semantics, HistoryUpdate), `docs/Configuration.md`
|
||||
(`ServerHistorian:Backend` + `SqlServer` subsection), `docs/security.md` (HistoryUpdate role/permission),
|
||||
`docs/Client.CLI.md`, `CLAUDE.md` Historian section, and the scadaproj `sister-project-history/otopcua.md`
|
||||
entry (ScadaBridge impact: none unless it issues HistoryUpdate; HistoryRead semantics unchanged on the
|
||||
gateway backend apart from honest Bad quality on Bad samples).
|
||||
@@ -0,0 +1,575 @@
|
||||
# Pluggable Historian Backends Implementation Plan
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:subagent-driven-development (recommended) or superpowers-extended-cc:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** Make the historian backend pluggable (`ServerHistorian:Backend`), carry typed values + real status end to end, ship a SQL Server backend, and add the OPC UA HistoryUpdate service.
|
||||
|
||||
**Architecture:** A Runtime-level `IHistorianBackendProvider` builds the five historian seams (data source, value writer, alarm writer, provisioning, updater); the Host picks one by name. The existing recorder/outbox/alarm S&F pipelines are unchanged in structure and now carry a typed `HistorianValue` + `uint StatusCode`. SQL Server stores samples in a `(TagId, SourceTimeUtc)`-clustered table and pushes aggregates down to T-SQL.
|
||||
|
||||
**Tech Stack:** .NET 10, Akka.NET, OPC Foundation UA .NET Standard, Microsoft.Data.SqlClient (already pinned centrally), FASTER log (existing), xUnit v3 + Shouldly.
|
||||
|
||||
**Spec:** `docs/plans/2026-10-09-pluggable-historian-design.md` — the binding authority. Read it before any task.
|
||||
|
||||
## Global Constraints
|
||||
|
||||
- **No new NuGet packages, no version changes.** `Directory.Packages.props` is not edited by any task. Adding `<PackageReference Include="Microsoft.Data.SqlClient" />` (already pinned) to a new project is allowed.
|
||||
- Recording stays **Primary-only** (#537 capture gate) and **at-least-once** with the #538 per-tag quarantine and all-or-nothing ack. Do not restructure `DrainOnceAsync`.
|
||||
- In any Akka actor handler: capture `IActorRef self = Self` (and `Sender`) **before** the first `await`; never touch `Self`/`Context` after an `await …ConfigureAwait(false)` (#491).
|
||||
- Test fakes for async seams used by actors must complete on a **pool thread** (`await Task.Delay(1).ConfigureAwait(false)`), never synchronously and never via `Task.Yield()`.
|
||||
- Every OPC UA status code a new class emits is a **named `const uint`** in an internal `*StatusCodes` class, and that assembly must be visible to `tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/StatusCodeParityTests.cs` (it scans its bin dir — add a ProjectReference).
|
||||
- Authorization changes need an **ALLOW** test (a deny-only test passes against a build that never consults the gate).
|
||||
- A regression test must be shown to fail with the production change reverted (falsification) — record the observation in the task report.
|
||||
- The gateway backend's observable behaviour (bytes sent for Good numeric samples, provisioning skips, read paths) does not change.
|
||||
- Commit messages end with `Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe`. Subagents never push and never dispatch subagents.
|
||||
- macOS zsh: quote globs; `tea` needs `</dev/null`.
|
||||
|
||||
**User decisions (already made):**
|
||||
- Approach A (provider seam over the existing five interfaces).
|
||||
- SQLite dropped; SQL Server is the first new backend; QuestDB is a later spec.
|
||||
- "Update" = OPC UA HistoryUpdate exposed to clients (Insert/Replace/Update/DeleteRaw/DeleteAtTime).
|
||||
- Typed values (Double/Int64/Boolean/String/DateTime + Empty), not numeric-only.
|
||||
- Aggregates pushed down per database (T-SQL), launch subset per spec §4.
|
||||
- Flag-off authorization uses a NEW `HistoryUpdate` role, not `WriteOperate`.
|
||||
- Gateway config keys stay top-level; new backends get subsections.
|
||||
- Testing target: the docker-dev `sql` container (`otopcua-dev-sql-1`, `localhost,14330`).
|
||||
|
||||
---
|
||||
|
||||
## Execution model: waves, models, and parallel safety
|
||||
|
||||
### Model per role
|
||||
|
||||
| Role | Model | Why |
|
||||
|---|---|---|
|
||||
| Plan review (before execution) | **fable** | user-requested |
|
||||
| Implementer, mechanical/standard tasks (T1, T2, T4, T5, T7, T8, T9, T11, T12, T13, T14) | **sonnet** | multi-file but fully specified |
|
||||
| Implementer, judgment-heavy tasks (T3 recorder actor, T6 node-manager HistoryUpdate + auth, T10 reads/aggregates) | **opus** | actor concurrency, auth semantics, SQL/Part 13 semantics |
|
||||
| Per-task reviewer | **sonnet** for sonnet tasks; **opus** for T3/T6/T10 | scale to risk |
|
||||
| Fix rounds 4–5 escalation | one tier up (sonnet→opus, opus→fable) | per SDD skill |
|
||||
| Final whole-branch review | **fable** | cross-task integration risk |
|
||||
| Live gate (T15) | controller (this session) | needs rig + judgement; never delegated |
|
||||
|
||||
Always pass `model` explicitly on every Agent dispatch.
|
||||
|
||||
### Waves
|
||||
|
||||
```
|
||||
Wave 1 (serial): T1 contracts + compile ripple [sonnet]
|
||||
Wave 2 (parallel): T2 outbox v2 ‖ T3 recorder ‖ T4 gateway provider ‖ T6 server HistoryUpdate ‖ T7 CLI ‖ T8 SQL skeleton+schema
|
||||
Wave 3 (parallel): T5 registry+Host ‖ T9 SQL writes ‖ T10 SQL reads ‖ T11 SQL updater ‖ T13 over-the-wire test
|
||||
Wave 4 (serial): T12 SQL provider + Host registration
|
||||
Wave 5 (serial): T14 docs → final fable review → T15 live gate (controller)
|
||||
```
|
||||
|
||||
Dependencies: T2,T3,T4,T6,T7,T8 ← T1. T5 ← T4. T9,T10,T11 ← T8. T13 ← T6. T12 ← T5,T9,T10,T11. T14 ← T12,T13. T15 ← T14 + final review.
|
||||
|
||||
### Parallel-safety rules (binding)
|
||||
|
||||
1. **One git worktree per parallel task**, created by the controller under `.worktrees/ph-T<N>` from the integration branch `feat/pluggable-historian` at the wave's start commit. Run `git -C <wt> log -1` and confirm the base SHA in the dispatch (worktrees can start stale — reset first).
|
||||
2. **File ownership.** A task edits only the files its "Files" list names. The shared hot files have exactly one owner per wave:
|
||||
|
||||
| Shared file | Owner |
|
||||
|---|---|
|
||||
| `ZB.MOM.WW.OtOpcUa.slnx` | T8 (adds both SqlServer projects) |
|
||||
| `src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs` | T5 (wave 3), then T12 (wave 4) |
|
||||
| `tests/Core/.../Core.Abstractions.Tests/*.csproj` | T8 |
|
||||
| `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs` | T6 only |
|
||||
| `.../Runtime/Historian/ContinuousHistorizationRecorder.cs` | T1 (ripple) then T3 only |
|
||||
| `.../Driver.Historian.SqlServer/*.csproj` | T8 only (T9–T11 add `.cs` files only; csproj uses default globbing + `EmbeddedResource` for `Schema/*.sql` set by T8) |
|
||||
|
||||
A task that discovers it needs an un-owned shared file stops and reports `NEEDS_CONTEXT` rather than editing it.
|
||||
3. **New-file-only parallelism in the SQL project.** T9/T10/T11 each create their own files and their own test files; none edits another's. Shared SQL helpers live in T8's `SqlHistorianConnection.cs`/`SampleRowMapper.cs`; if a wave-3 task needs a helper change, it adds a private helper in its own file instead.
|
||||
4. **Merge order** at wave end: the controller merges worktree branches into `feat/pluggable-historian` in ascending task number, running `dotnet build ZB.MOM.WW.OtOpcUa.slnx` after each merge and the touched test projects after the last. A conflict ⇒ stop and resolve on the integration branch, never inside a worktree.
|
||||
5. **Integration-test DBs are per-test-run** (`OtOpcUaHistorian_T_<guid>`), so parallel T9/T10/T11 suites against the same `sql` container cannot collide.
|
||||
6. Subagents do not push, merge, or dispatch subagents. Only the controller merges.
|
||||
|
||||
### Common commands
|
||||
|
||||
```bash
|
||||
dotnet build ZB.MOM.WW.OtOpcUa.slnx
|
||||
dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests
|
||||
dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests --filter "FullyQualifiedName~Historian"
|
||||
dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests
|
||||
export HISTORIAN_SQL_TEST_CONNECTION='Server=localhost,14330;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True;Encrypt=False'
|
||||
dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests
|
||||
```
|
||||
|
||||
The rig `sql` container must be up: `docker ps --filter name=otopcua-dev-sql-1`. If down: `cd docker-dev && docker compose up -d sql`.
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Typed historian contracts + compile ripple
|
||||
|
||||
**Goal:** Introduce `HistorianValue`, `HistorianValueKind`, `HistoryUpdateMode`, `IHistorianDataUpdater`, `IHistorianBackendProvider`, change `HistorizationValue`/`HistorizationOutboxEntry` to typed+status, and keep the whole solution compiling with **no behaviour change**.
|
||||
|
||||
**Files:**
|
||||
- Create: `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistorianValue.cs`
|
||||
- Create: `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataUpdater.cs`
|
||||
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianBackendProvider.cs`
|
||||
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianValueWriter.cs` (`HistorizationValue`)
|
||||
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistorizationOutboxEntry.cs`
|
||||
- Modify (ripple only): `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs` (lines ~548 entry construction, ~692 drain mapping), `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/GatewayHistorianValueWriter.cs`, `.../Recorder/HistorizationOutboxEntrySerializer.cs`, and every test that constructs these records (find with `grep -rn "HistorizationOutboxEntry(\|HistorizationValue(" src tests --include='*.cs'`).
|
||||
- Test: `tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianValueTests.cs`
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] `HistorianValue.TryFromObject` implements exactly the spec §3 table; tests cover every row incl. `ulong` above `long.MaxValue` ⇒ Double, `DateTime` local ⇒ UTC ticks, `null` ⇒ Empty, `int[]` ⇒ false.
|
||||
- [ ] `TryGetDouble` true for Double/Int64/Boolean only; `ToObject` round-trips each kind (`DateTime` returns `DateTimeKind.Utc`).
|
||||
- [ ] Solution builds; all pre-existing historian tests pass unchanged in assertion content (only constructor call sites updated).
|
||||
- [ ] Ripple preserves behaviour: recorder still coerces numerics only (via `TryFromObject` + `TryGetDouble`, building `HistorianValue.FromDouble(numeric)`) and still stamps Good (`StatusCode = 0`); serializer still writes the legacy layout using `Value.Number` and quality `192`; gateway writer maps `Value.Number` and quality `192`.
|
||||
|
||||
**Verify:** `dotnet build ZB.MOM.WW.OtOpcUa.slnx && dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests --filter "FullyQualifiedName~HistorianValue" && dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests && dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests --filter "FullyQualifiedName~Historiz"` → all green.
|
||||
|
||||
**Steps:**
|
||||
|
||||
- [ ] **Step 1: Write `HistorianValueTests` (failing — type does not exist).** Cases:
|
||||
|
||||
```csharp
|
||||
[Theory]
|
||||
[InlineData(1.5d, HistorianValueKind.Double)]
|
||||
[InlineData(1.5f, HistorianValueKind.Double)]
|
||||
[InlineData(7, HistorianValueKind.Int64)]
|
||||
[InlineData((short)7, HistorianValueKind.Int64)]
|
||||
[InlineData((byte)7, HistorianValueKind.Int64)]
|
||||
[InlineData(7u, HistorianValueKind.Int64)]
|
||||
[InlineData(7L, HistorianValueKind.Int64)]
|
||||
[InlineData(true, HistorianValueKind.Boolean)]
|
||||
[InlineData("x", HistorianValueKind.String)]
|
||||
public void Maps_clr_values_to_kinds(object input, HistorianValueKind expected)
|
||||
{
|
||||
HistorianValue.TryFromObject(input, out var v).ShouldBeTrue();
|
||||
v.Kind.ShouldBe(expected);
|
||||
}
|
||||
// plus: decimal 2.5m ⇒ Double 2.5; ulong.MaxValue ⇒ Double; (ulong)5 ⇒ Int64 5;
|
||||
// null ⇒ Empty (true); new[]{1,2} ⇒ false; new object() ⇒ false;
|
||||
// DateTime(2026,1,1,0,0,0,Local) ⇒ DateTime kind, Integer == ToUniversalTime().Ticks, ToObject() Kind Utc;
|
||||
// FromBoolean(true): Integer 1, Number 1, TryGetDouble ⇒ 1; FromString("a").TryGetDouble ⇒ false; Empty.ToObject() ⇒ null.
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Run, confirm compile failure.**
|
||||
- [ ] **Step 3: Implement `HistorianValue.cs`:**
|
||||
|
||||
```csharp
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
/// <summary>The value kinds a historian backend can store. Persisted as a byte (SQL ValueKind, outbox v2) — never renumber.</summary>
|
||||
public enum HistorianValueKind : byte { Empty = 0, Double = 1, Int64 = 2, Boolean = 3, String = 4, DateTime = 5 }
|
||||
|
||||
/// <summary>A typed historized value. Number carries Double (and the double view of Int64/Boolean);
|
||||
/// Integer carries Int64, Boolean (1/0) and DateTime (UTC ticks); Text carries String.</summary>
|
||||
public readonly record struct HistorianValue(HistorianValueKind Kind, double Number, long Integer, string? Text)
|
||||
{
|
||||
public static HistorianValue Empty => default;
|
||||
public static HistorianValue FromDouble(double v) => new(HistorianValueKind.Double, v, 0, null);
|
||||
public static HistorianValue FromInt64(long v) => new(HistorianValueKind.Int64, v, v, null);
|
||||
public static HistorianValue FromBoolean(bool v) => new(HistorianValueKind.Boolean, v ? 1d : 0d, v ? 1 : 0, null);
|
||||
public static HistorianValue FromString(string v) => new(HistorianValueKind.String, 0d, 0, v ?? throw new ArgumentNullException(nameof(v)));
|
||||
public static HistorianValue FromDateTime(DateTime v) =>
|
||||
new(HistorianValueKind.DateTime, 0d, (v.Kind == DateTimeKind.Local ? v.ToUniversalTime() : DateTime.SpecifyKind(v, DateTimeKind.Utc)).Ticks, null);
|
||||
|
||||
public static bool TryFromObject(object? value, out HistorianValue result)
|
||||
{
|
||||
switch (value)
|
||||
{
|
||||
case null: result = Empty; return true;
|
||||
case double d: result = FromDouble(d); return true;
|
||||
case float f: result = FromDouble(f); return true;
|
||||
case decimal m: result = FromDouble((double)m); return true;
|
||||
case sbyte sb: result = FromInt64(sb); return true;
|
||||
case byte b: result = FromInt64(b); return true;
|
||||
case short s: result = FromInt64(s); return true;
|
||||
case ushort us: result = FromInt64(us); return true;
|
||||
case int i: result = FromInt64(i); return true;
|
||||
case uint ui: result = FromInt64(ui); return true;
|
||||
case long l: result = FromInt64(l); return true;
|
||||
case ulong ul: result = ul <= long.MaxValue ? FromInt64((long)ul) : FromDouble(ul); return true;
|
||||
case bool bo: result = FromBoolean(bo); return true;
|
||||
case string str: result = FromString(str); return true;
|
||||
case DateTime dt: result = FromDateTime(dt); return true;
|
||||
default: result = Empty; return false; // arrays, references, structures ⇒ not historizable
|
||||
}
|
||||
}
|
||||
|
||||
public bool TryGetDouble(out double value)
|
||||
{
|
||||
switch (Kind)
|
||||
{
|
||||
case HistorianValueKind.Double:
|
||||
case HistorianValueKind.Int64:
|
||||
case HistorianValueKind.Boolean: value = Number; return true;
|
||||
default: value = 0d; return false;
|
||||
}
|
||||
}
|
||||
|
||||
public object? ToObject() => Kind switch
|
||||
{
|
||||
HistorianValueKind.Double => Number,
|
||||
HistorianValueKind.Int64 => Integer,
|
||||
HistorianValueKind.Boolean => Integer != 0,
|
||||
HistorianValueKind.String => Text,
|
||||
HistorianValueKind.DateTime => new DateTime(Integer, DateTimeKind.Utc),
|
||||
_ => null,
|
||||
};
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: `IHistorianDataUpdater.cs`** — exactly the spec §4 "Updates" interface + `HistoryUpdateMode` enum, with XML docs naming the per-item status codes.
|
||||
- [ ] **Step 5: `IHistorianBackendProvider.cs`** in `ZB.MOM.WW.OtOpcUa.Runtime.Historian` — exactly spec §2.
|
||||
- [ ] **Step 6: Change the records:**
|
||||
|
||||
```csharp
|
||||
public readonly record struct HistorizationValue(DateTime? TimestampUtc, HistorianValue Value, uint StatusCode);
|
||||
public sealed record HistorizationOutboxEntry(Guid Id, string Tag, HistorianValue Value, uint StatusCode, DateTime TimestampUtc);
|
||||
```
|
||||
|
||||
- [ ] **Step 7: Ripple, behaviour-preserving.** Recorder line ~527: replace `TryCoerceToDouble(msg.Value, out double numeric)` with `HistorianValue.TryFromObject(msg.Value, out var hv) && hv.TryGetDouble(out double numeric)` (same drop semantics — strings/null/DateTime still dropped as non-numeric); entry = `new HistorizationOutboxEntry(Guid.NewGuid(), historianName, HistorianValue.FromDouble(numeric), 0u /*Good*/, timestampUtc)`; delete `TryCoerceToDouble`. Drain line ~692: `new HistorizationValue(e.TimestampUtc, e.Value, e.StatusCode)`. Serializer: write `entry.Value.Number` and `(ushort)192`; read back as `FromDouble(value)`, status `0`. Gateway writer: `NumericValue = value.Value.Number, Quality = 192`. Keep `GoodQuality` const for now (T3 removes it). Update test call sites.
|
||||
- [ ] **Step 8: Build + run Verify.** Commit: `feat(historian): typed HistorianValue contracts, IHistorianDataUpdater, IHistorianBackendProvider (no behaviour change)`.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistorianValue.cs","src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataUpdater.cs","src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianBackendProvider.cs","src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianValueWriter.cs","src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistorizationOutboxEntry.cs","src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/GatewayHistorianValueWriter.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/HistorizationOutboxEntrySerializer.cs"],"acceptanceCriteria":["TryFromObject matches spec table","solution builds, existing historian tests green","no behaviour change"],"modelTier":"standard","model":"sonnet","wave":1}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: Outbox format v2 with legacy read
|
||||
|
||||
**Goal:** Serialize every `HistorianValueKind` + `uint StatusCode`; read pre-upgrade entries; drop (and count) unreadable entries instead of throwing.
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/HistorizationOutboxEntrySerializer.cs`
|
||||
- Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs` (only the deserialize call site in Peek: skip+count unreadable entries; expose `long UnreadableCount`)
|
||||
- Test: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/HistorizationOutboxEntrySerializerV2Tests.cs`
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] v2 layout exactly: `"HV2\0"`(4) `ver=2`(1) `id`(16) `tagLen`(i32 LE) `tag`(UTF-8) `kind`(1) payload `status`(u32 LE) `ts`(i64 LE `ToBinary`). Payload: Empty none; Double f64; Int64/Boolean/DateTime i64; String i32 len + UTF-8.
|
||||
- [ ] Round-trip test per kind incl. empty string, a 70-KB string, `NaN`, `double.NegativeInfinity`, `long.MinValue`, status `0x80310000`.
|
||||
- [ ] **Pinned legacy bytes:** a hex literal produced by today's serializer (generate it once from the T1-era `Serialize` for `Id=00000000-0000-0000-0000-000000000001, Tag="t", Value=42.5, Quality=192, ts=2026-01-01T00:00:00Z` and paste the bytes into the test) deserializes to `Double 42.5`, status `0`.
|
||||
- [ ] `TryDeserialize` returns false (never throws) for: 3 random bytes, magic with ver=9, magic with truncated payload, legacy-looking bytes whose length disagrees with `38+tagLen`. `FasterLogHistorizationOutbox` skips such an entry on peek, advances past it (it must be removable so it can't pin the head — truncate through its id or log-address as the outbox already does for acked entries), and increments `UnreadableCount`; a test proves a corrupt entry between two good ones does not block the second.
|
||||
- [ ] Meter: `ContinuousHistorizationMetrics` gets an observable `otopcua.historian.outbox.unreadable` bound alongside the existing outbox gauges — **only if** `BindOutbox` lives in an un-owned file; otherwise report the counter on the outbox and note it for T5. (Owned file check: `ContinuousHistorizationMetrics.cs` is NOT owned by T2 — expose the property only.)
|
||||
|
||||
**Verify:** `dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests --filter "FullyQualifiedName~Outbox"` → green; falsification: with the legacy branch deleted the pinned-bytes test fails.
|
||||
|
||||
**Steps:** (1) capture the legacy hex from the current serializer via a throwaway test print, paste into the new test; (2) write failing v2 tests; (3) implement `Serialize` (v2 always) + `TryDeserialize(ReadOnlySpan<byte>, out HistorizationOutboxEntry?)` (v2 by magic, else legacy by exact length, else false) and keep a throwing `Deserialize` wrapper only if other callers need it; (4) wire skip+count in the outbox peek; (5) green; commit `feat(historian): outbox format v2 carries typed values + status; legacy entries still read`.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/HistorizationOutboxEntrySerializer.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs"],"acceptanceCriteria":["v2 layout","pinned legacy bytes read","corrupt entry skipped and counted, never pins head"],"modelTier":"standard","model":"sonnet","wave":2}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: Recorder captures typed values with real status, filtered by backend kinds
|
||||
|
||||
**Goal:** The recorder records `HistorianValue` + `msg.StatusCode` and drops kinds the backend does not support.
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs`
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs` **only** the recorder `Props` construction (~line 485–520) to pass `supportedKinds` — resolve `IHistorianBackendProvider` optionally; absent ⇒ `{Double, Int64, Boolean}` (today's behaviour).
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTypedValueTests.cs`
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] New ctor/Props parameter `IReadOnlySet<HistorianValueKind> supportedKinds` (default = `{Double, Int64, Boolean}` for existing overloads so other tests compile unchanged).
|
||||
- [ ] Capture: `TryFromObject` false ⇒ drop + `_droppedUnsupportedKind++`; kind ∉ supportedKinds ⇒ same; otherwise entry carries the typed value and `msg.StatusCode`.
|
||||
- [ ] Status field in the recorder status record renamed `DroppedNonNumeric` → `DroppedUnsupportedKind`; any OTel instrument name stays unchanged (dashboards). `GoodQuality` const and its doc paragraph removed; class remarks updated (the mux DOES carry StatusCode — `VirtualTagActor.DependencyValueChanged`).
|
||||
- [ ] Tests (fake writer completes on a pool thread): a `Bad (0x80000000)` numeric sample is written with status `0x80000000`; a string sample is written when kinds include String and dropped+counted when not; `null` with status `BadNoCommunication` is written as Empty when kinds include Empty, dropped otherwise; an `int[]` is always dropped; existing #537 capture-gate and #538 quarantine tests still pass untouched.
|
||||
- [ ] Falsification recorded: reverting to the hard-coded Good makes the Bad-status test fail.
|
||||
|
||||
**Verify:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests --filter "FullyQualifiedName~ContinuousHistorization"` → green.
|
||||
|
||||
**Steps:** failing tests → implement → green → grep for `DroppedNonNumeric` across src/tests and update readers (AdminUI/metrics) **only within Runtime**; if a reader exists outside the owned files, report it for T5 instead of editing. Commit `fix(historian): record the real status code and typed values; backend-supported kinds gate capture`.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs","src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs"],"acceptanceCriteria":["real status recorded","kind filter","#537/#538 unchanged","pool-thread fake"],"modelTier":"frontier","model":"opus","wave":2}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 4: Gateway provider + reverse quality mapping
|
||||
|
||||
**Goal:** `GatewayHistorianProvider : IHistorianBackendProvider`; the gateway writer maps typed value + status to the gateway's numeric + DA quality.
|
||||
|
||||
**Files:**
|
||||
- Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianProvider.cs`
|
||||
- Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/GatewayQualityMapper.cs` (add `ToDaQuality(uint)`)
|
||||
- Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/GatewayHistorianValueWriter.cs`
|
||||
- Test: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianProviderTests.cs`, `.../Mapping/GatewayQualityMapperReverseTests.cs`
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] Provider: `Name = "Gateway"`, `SupportedKinds = {Double, Int64, Boolean}`, `CreateUpdater` returns `null`; the four `Create*` delegate to the existing `GatewayHistorian.Create…` (data source/alarm writer/provisioner) and build `GatewayHistorianValueWriter` exactly as `Program.cs` does today (`HistorianGatewayClientAdapter.Create(options, loggerFactory)`). Options come from the ctor (`ServerHistorianOptions`).
|
||||
- [ ] `ToDaQuality(Map(q)) == q` for every explicit byte in `Map`'s table; unknown status falls back by severity bits (`0x00…`⇒192, `0x40…`⇒64, `0x80…`⇒0).
|
||||
- [ ] Writer: a value with `TryGetDouble` false is skipped (debug log, no throw); Good Double 42.5 produces the same `HistorianLiveValue` (`NumericValue=42.5, Quality=192`) as before T1 — asserted in a test.
|
||||
|
||||
**Verify:** `dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests` → green.
|
||||
|
||||
**Steps:** tests first → implement → green → commit `feat(historian): GatewayHistorianProvider; status→DA quality reverse mapping`.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianProvider.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/GatewayQualityMapper.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/GatewayHistorianValueWriter.cs"],"acceptanceCriteria":["provider shape","reverse mapping inverse","Good bytes unchanged"],"modelTier":"standard","model":"sonnet","wave":2}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 5: Backend registry, config, validators, Host wiring
|
||||
|
||||
**Goal:** `ServerHistorian:Backend` selects a provider; Program.cs builds every historian seam through it.
|
||||
|
||||
**Files:**
|
||||
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs`
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs` (add `Backend` string, default `"Gateway"`)
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs`
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs` (lines ~215–318 only)
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs`, extend `ServerHistorianOptionsValidatorTests`
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] Registry: `static IHistorianBackendProvider Resolve(ServerHistorianOptions o, IConfiguration cfg)`; knows `"Gateway"` now (T12 adds `"SqlServer"`); case-insensitive; unknown ⇒ `OptionsValidationException` naming the value and listing valid names. The registry also exposes `IReadOnlyCollection<string> KnownBackends` used by the validator.
|
||||
- [ ] Validator: unknown backend fails; `Endpoint` required only when backend is Gateway; existing tests unchanged in intent.
|
||||
- [ ] Program.cs: resolve the provider once; register it as a singleton `IHistorianBackendProvider`; replace the 4 `GatewayHistorian.Create…` lambdas and the inline `GatewayHistorianValueWriter` with `provider.Create…(sp)`; register `IHistorianDataUpdater` via `provider.CreateUpdater(sp)` **only when non-null** (`TryAddSingleton`-style factory is not possible for a null — register conditionally by calling it once at registration time is NOT allowed (needs sp); instead register a small `HistorianUpdaterHolder(IHistorianDataUpdater? Updater)` singleton, and have T6's hosted-service hook read `holder.Updater`). The alarm-only warning text names `serverHistorianOptions.Backend`.
|
||||
- [ ] Bind the outbox unreadable gauge from T2 in `ContinuousHistorizationMetrics.BindOutbox` if T2 reported it (this task owns that file this wave).
|
||||
- [ ] With `Backend` absent the Host's DI graph resolves the same concrete types as before (test: build the service collection with a Gateway config and assert `IHistorianDataSource` resolves to `GatewayHistorianDataSource`).
|
||||
|
||||
**Verify:** `dotnet build ZB.MOM.WW.OtOpcUa.slnx && dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests --filter "FullyQualifiedName~Historian"` → green.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs","src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs","src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs","src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs"],"acceptanceCriteria":["unknown backend fails boot","Gateway default resolves identical types","updater holder registered"],"modelTier":"standard","model":"sonnet","wave":3}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 6: HistoryUpdate service in the node manager + authorization
|
||||
|
||||
**Goal:** Clients can HistoryUpdate historized nodes when the backend supplies an updater, gated by ACL (flag on) or the new `HistoryUpdate` role (flag off).
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs`
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/Security/OpcUaDataPlaneRoles.cs` (`HistoryUpdate = "HistoryUpdate"`)
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/ApiKeyInterimRoleMapper.cs`
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/OtOpcUaServerHostedService.cs` (assign `HistorianDataUpdater` from an optional `HistorianUpdaterHolder` — define `HistorianUpdaterHolder` in `Runtime/Historian/HistorianUpdaterHolder.cs` in THIS task: `public sealed record HistorianUpdaterHolder(IHistorianDataUpdater? Updater) { public static readonly HistorianUpdaterHolder None = new((IHistorianDataUpdater?)null); }`, resolve with `GetService` ⇒ None when absent)
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Clusters/AclEdit.razor` (add `NodePermissions.HistoryUpdate` to the list at ~line 163–166 with label "History update")
|
||||
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Enums/NodePermissions.cs` comment only (remove "until the HistoryUpdate service surface is implemented"; still in no bundle)
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/History/HistoryUpdateTests.cs`, `tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/OpcUa/ApiKeyInterimRoleMapperTests.cs` (extend)
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] Before writing, read the pinned SDK's `CustomNodeManager2` and confirm the exact protected signatures for `HistoryUpdateData`, `HistoryDeleteRawModified`, `HistoryDeleteAtTime` (and how `HistoryUpdate` dispatches by details type); record them in the report. Use DeepWiki (`OPCFoundation/UA-.NETStandard`) if the source is not on disk.
|
||||
- [ ] `public IHistorianDataUpdater? HistorianDataUpdater { get; set; }` (volatile backing field, like `HistorianDataSource`).
|
||||
- [ ] `UpdateDataDetails`: `PerformUpdateType.Insert/Replace/Update` ⇒ `HistoryUpdateMode`; `Remove` ⇒ `BadHistoryOperationUnsupported`. Values: `DataValue` ⇒ `HistorizationValue(SourceTimestamp, TryFromObject(Value.Value) or BadTypeMismatch item, StatusCode.Code)`. Per-item results returned in `OperationResults` in order.
|
||||
- [ ] `DeleteRawModifiedDetails` with `IsDeleteModified=false` ⇒ `DeleteRawAsync`; `true` ⇒ unsupported. `DeleteAtTimeDetails` ⇒ `DeleteAtTimeAsync`.
|
||||
- [ ] Not historized / no updater ⇒ `BadHistoryOperationUnsupported`. Event/structure details ⇒ unsupported.
|
||||
- [ ] Authorization helper `AuthorizeHistoryUpdate` mirroring `AuthorizeHistoryRead` + the WriteOperate pattern at ~line 2024–2033: flag on ⇒ `_authorizer.Authorize(identity, OpcUaOperation.HistoryUpdate, nodeId, realm)`; flag off ⇒ `AuthorizeKeyUsable(...)` AND role `HistoryUpdate`. Deny ⇒ `BadUserAccessDenied`.
|
||||
- [ ] Batch limiter + deadline shared with HistoryRead (`WithHistoryReadBatchLimiter`, `CreateDeadline`).
|
||||
- [ ] `ComposeAccessLevel(writable, historized, historyWritable)` ORs `AccessLevels.HistoryWrite` when `historized && historyWritable`; `historyWritable = HistorianDataUpdater is not null` at materialise time. Raw + UNS mirrors both get it. (Setting the updater after nodes exist does not retro-update AccessLevel — document; the hosted service assigns it before the first deploy applies, verify the ordering and state it.)
|
||||
- [ ] `ApiKeyInterimRoleMapper` adds `OpcUaDataPlaneRoles.HistoryUpdate` for a Cluster-scoped grant with `NodePermissions.HistoryUpdate`; a WriteOperate-only grant does NOT confer it.
|
||||
- [ ] Tests (fake updater + fake authorizer, following the existing node-manager test harness for HistoryRead): **ALLOW flag-on with no roles + scoped grant ⇒ updater called and `GoodEntryInserted` returned**; flag-off with `HistoryUpdate` role ⇒ allowed; flag-off with only `WriteOperate` ⇒ `BadUserAccessDenied` and updater NOT called; expired key ⇒ denied; no updater ⇒ unsupported; per-item codes passed through in order; UNS NodeId resolves to the same tagname as raw.
|
||||
- [ ] Falsification recorded for the ALLOW test (break the flag-on branch ⇒ it fails).
|
||||
|
||||
**Verify:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests --filter "FullyQualifiedName~HistoryUpdate" && dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests --filter "FullyQualifiedName~ApiKeyInterimRoleMapper"` → green.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs","src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/Security/OpcUaDataPlaneRoles.cs","src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/ApiKeyInterimRoleMapper.cs","src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/OtOpcUaServerHostedService.cs","src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianUpdaterHolder.cs","src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Clusters/AclEdit.razor","src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Enums/NodePermissions.cs"],"acceptanceCriteria":["SDK signatures verified","ALLOW test","WriteOperate does not confer HistoryUpdate","HistoryWrite access bit"],"modelTier":"frontier","model":"opus","wave":2}
|
||||
```
|
||||
|
||||
Note: T5 references `HistorianUpdaterHolder` — T5 runs in wave 3 after T6 merges, so the type exists.
|
||||
|
||||
---
|
||||
|
||||
### Task 7: Client.CLI `historyupdate` command
|
||||
|
||||
**Goal:** A CLI command to drive HistoryUpdate for manual + live-gate testing.
|
||||
|
||||
**Files:**
|
||||
- Create: `src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI/Commands/HistoryUpdateCommand.cs` (follow the existing `historyread` command's structure, options parsing, session/auth handling exactly)
|
||||
- Modify: `docs/Client.CLI.md` (new section)
|
||||
- Test: the Client.CLI test project if one exists for commands (`ls tests/Client`); follow its pattern for argument parsing tests.
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] Options per spec §6; `--mode` values `insert|replace|update|delete-raw|delete-at-time`; `--value` parsed against `--type double|int64|bool|string|datetime` (default double); repeated `--time/--value` pairs; validation errors exit non-zero with a message.
|
||||
- [ ] Sends `UpdateDataDetails` / `DeleteRawModifiedDetails(IsDeleteModified=false)` / `DeleteAtTimeDetails` via `Session.HistoryUpdate`; prints `index time StatusCodeName` per item plus the overall service result.
|
||||
- [ ] Argument-parsing unit tests for each mode (no server needed).
|
||||
|
||||
**Verify:** `dotnet build src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI && dotnet test tests/Client --filter "FullyQualifiedName~HistoryUpdate"` → green.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI/Commands/HistoryUpdateCommand.cs","docs/Client.CLI.md"],"acceptanceCriteria":["all five modes","arg parsing tests"],"modelTier":"standard","model":"sonnet","wave":2}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 8: SQL Server backend project skeleton, schema, bootstrapper, test fixture
|
||||
|
||||
**Goal:** New projects, embedded DDL, idempotent locked bootstrap, options + validation, shared connection/mapping helpers, and the env-gated integration fixture.
|
||||
|
||||
**Files:**
|
||||
- Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj` (refs: Core.Abstractions, Core.AlarmHistorian, Runtime; `<PackageReference Include="Microsoft.Data.SqlClient" />`, `Microsoft.Extensions.Logging.Abstractions`; `<EmbeddedResource Include="Schema\*.sql" />`; `InternalsVisibleTo` both test projects)
|
||||
- Create: `.../SqlServerHistorianOptions.cs` (+ `Validate()` returning errors per spec §2), `.../SqlHistorianStatusCodes.cs`, `.../SqlHistorianConnection.cs` (open connection, schema-qualify helper `Q(string obj)` ⇒ `[{schema}].[{obj}]`, transient-error classifier: 1205, -2, 40613, 40197, 40501, 49918, 4060, 233, 10053, 10054, 10060, 11001, SqlException with `IsTransient`), `.../SampleRowMapper.cs` (HistorianValue ⇄ columns per spec §4 + `DataTable` builders for `SampleRow` and `TimeRow` TVPs), `.../SqlServerSchemaBootstrapper.cs`, `.../Schema/001_initial.sql`
|
||||
- Create: `docs/historian/sqlserver/001_initial.sql` (copy of the embedded script)
|
||||
- Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/` (unit) and `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/` with `SqlHistorianFixture.cs`
|
||||
- Modify: `ZB.MOM.WW.OtOpcUa.slnx` (4 new projects), `tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests.csproj` (ProjectReference to the new src project so `StatusCodeParityTests` scans `SqlHistorianStatusCodes`)
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] `001_initial.sql` is the spec §4 DDL with `{s}` replaced by a `$(Schema)` token that the bootstrapper substitutes after validating the identifier regex; every statement guarded (`IF SCHEMA_ID(...) IS NULL`, `IF OBJECT_ID(...) IS NULL`, `IF TYPE_ID(...) IS NULL`) so a re-run is a no-op; ends with `INSERT SchemaVersion(1, SYSUTCDATETIME())` guarded by `IF NOT EXISTS`. Split on `^GO$` lines.
|
||||
- [ ] `SqlServerSchemaBootstrapper.EnsureAsync(ct)`: lazy, once per instance (a `SemaphoreSlim` + `_ready` flag; failure NOT latched), `sp_getapplock` exclusive session lock with a 30 s timeout around the apply; `AutoCreateSchema=false` ⇒ only checks `MAX(Version) >= 1` and throws `HistorianSchemaMissingException("…missing version 1…")`.
|
||||
- [ ] `SqlHistorianStatusCodes` contains every code the spec names (`Good, GoodNoData, GoodEntryInserted, GoodEntryReplaced, BadNoData, BadEntryExists, BadNoEntryExists, BadInvalidTimestamp, BadTypeMismatch, BadAggregateInvalidInputs, BadHistoryOperationUnsupported, UncertainDataSubNormal`) as named `const uint` with values copied from `Opc.Ua.StatusCodes` — the parity test passes and lists this class.
|
||||
- [ ] Historian info-bit constants (`InfoTypeDataValue = 0x00000400`, `HistorianCalculated = 0x1`, `HistorianInterpolated = 0x2`) in a separate `HistorianInfoBits` static class, with a unit test asserting `new Opc.Ua.StatusCode(Good).SetAggregateBits(AggregateBits.Calculated).Code == (Good | InfoTypeDataValue | HistorianCalculated)` (adjust to the SDK's actual helper — the test must compare against the SDK, not against this plan).
|
||||
- [ ] `SqlHistorianFixture` (xUnit `IAsyncLifetime`, collection fixture): reads `HISTORIAN_SQL_TEST_CONNECTION`; absent ⇒ `SkipReason` and tests `Assert.Skip(SkipReason)` (follow `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlPollServerFixture.cs`); present ⇒ `CREATE DATABASE OtOpcUaHistorian_T_<guid:N>`, exposes `ConnectionString` for it, `DROP DATABASE` (after `ALTER … SET SINGLE_USER WITH ROLLBACK IMMEDIATE`) on dispose.
|
||||
- [ ] Integration tests: bootstrap twice ⇒ second is a no-op and `SchemaVersion` has one row; two bootstrappers concurrently ⇒ both succeed, one row; `AutoCreateSchema=false` on an empty DB ⇒ `HistorianSchemaMissingException`; invalid schema name rejected by `Validate()` before any SQL runs (unit).
|
||||
- [ ] `SampleRowMapper` unit tests: every kind round-trips through the column triple.
|
||||
|
||||
**Verify:** `dotnet build ZB.MOM.WW.OtOpcUa.slnx && dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests && dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests && dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests --filter "FullyQualifiedName~StatusCodeParity"` → green (integration with the env var set; also run once WITHOUT it and confirm all skip).
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/","tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/","tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/","ZB.MOM.WW.OtOpcUa.slnx","docs/historian/sqlserver/001_initial.sql"],"acceptanceCriteria":["idempotent DDL","applock bootstrap","parity-guarded status consts","env-gated throwaway DB fixture"],"modelTier":"standard","model":"sonnet","wave":2}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 9: SQL Server writes — value writer, provisioning, alarm writer
|
||||
|
||||
**Goal:** Live values, tag provisioning and alarm events persist idempotently.
|
||||
|
||||
**Files (create only):**
|
||||
- `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagCatalog.cs` (TagId cache + `GetOrCreateAsync(name, kindHint)` via `MERGE … OUTPUT inserted.TagId`)
|
||||
- `.../SqlServerHistorianValueWriter.cs`, `.../SqlServerTagProvisioner.cs`, `.../SqlServerAlarmHistorianWriter.cs`
|
||||
- Tests: `tests/.../SqlServer.IntegrationTests/WriteTests.cs`, `ProvisioningTests.cs`, `AlarmWriterTests.cs`; unit `tests/.../SqlServer.Tests/AlarmEventIdTests.cs`
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] Every public entry point calls `bootstrapper.EnsureAsync` first.
|
||||
- [ ] Value writer: TVP insert-if-absent (`INSERT INTO Sample(...) SELECT @tagId, r.* , SYSUTCDATETIME() FROM @rows r WHERE NOT EXISTS (SELECT 1 FROM Sample s WITH (UPDLOCK, HOLDLOCK) WHERE s.TagId=@tagId AND s.SourceTimeUtc=r.SourceTimeUtc)`); within-batch duplicate timestamps collapse to the first (de-dup in C# before building the TVP — TVP PK would reject them); null timestamp ⇒ `DateTime.UtcNow`; returns false (never throws) on SQL errors; logs category only.
|
||||
- [ ] Provisioning: one `MERGE` per batch upserting Name/DataType/EU/Description; returns `(Requested, Ensured, Skipped:0, Failed)`; never throws.
|
||||
- [ ] Alarm writer: EventId = lowercase hex SHA-256 of `AlarmId|EventKind|TimestampUtc.Ticks|User|Comment|Message` (UTF-8, `|` separator, null ⇒ empty); insert-if-absent; per-event outcome per spec (transient classifier from T8).
|
||||
- [ ] Integration: each kind written then read back by direct SQL equals input; writing the same batch twice leaves row count unchanged and returns true; a 10 000-value batch completes < 5 s on the rig; provisioning upsert changes DataType on re-run; same alarm event twice ⇒ one row, both `Ack`; an over-long `AlarmTypeName` (> 100) ⇒ `PermanentFail`.
|
||||
|
||||
**Verify:** `dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests --filter "FullyQualifiedName~Write|FullyQualifiedName~Provisioning|FullyQualifiedName~AlarmWriter"` → green.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagCatalog.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs"],"acceptanceCriteria":["idempotent redelivery","never throws","deterministic EventId"],"modelTier":"standard","model":"sonnet","wave":3}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 10: SQL Server reads — raw, at-time, events, processed (push-down)
|
||||
|
||||
**Goal:** `SqlServerHistorianDataSource : IHistorianDataSource` per spec §4 "Reads".
|
||||
|
||||
**Files (create only):** `.../SqlServerHistorianDataSource.cs`, `.../SqlAggregateQueries.cs` (one method per aggregate family returning the SQL text), tests `tests/.../SqlServer.IntegrationTests/RawReadTests.cs`, `AtTimeReadTests.cs`, `EventReadTests.cs`, `ProcessedReadTests.cs`.
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] Tag lookup is read-only (`SELECT TagId FROM Tag WHERE Name=@n`, cached on hit only) — reads never create tags.
|
||||
- [ ] Raw forward/reverse/limit exactly per spec; `DataValueSnapshot` built with `Value.ToObject()`, `StatusCode`, source + server timestamps (inspect `DataValueSnapshot`'s ctor first).
|
||||
- [ ] AtTime: single statement with `TimeRow` TVP and two `OUTER APPLY (SELECT TOP 1 …)`; interpolation/stepping/no-prior/bad-neighbour rules per spec; status built from `SqlHistorianStatusCodes` + `HistorianInfoBits`.
|
||||
- [ ] Processed: generates every interval in `[start,end)` (empty ⇒ `BadNoData`), the aggregate table in spec §4, Calculated bit on every result, Uncertain when any Bad present, `BadAggregateInvalidInputs` for numeric aggregates on String/DateTime tags, `HistoryAggregateNotSupportedException` for the rest. `interval <= 0` ⇒ single interval `[start,end)`.
|
||||
- [ ] Reverse processed (`end < start`) ⇒ swap and return descending (document).
|
||||
- [ ] `GetHealthSnapshot()` tracks totals/consecutive failures/last error (the bootstrap error surfaces here).
|
||||
- [ ] Integration fixture data: tag `A` with samples at t0+0s=10, +10s=20, +20s=Bad(30), +30s=40, +40s=50 (Double); tag `S` strings; `B` booleans. Hand-computed expectations for each supported aggregate over 20 s intervals from t0 — write them as literal expected values in the test (e.g. interval [t0,t0+20s): Average 15, Min 10, Max 20, Total 30, Count 2, Start 10, End 20, Delta 10, Range 10, PercentGood 100; interval [t0+20s,t0+40s): usable {40}, Average 40 with Uncertain status, PercentBad 50).
|
||||
- [ ] Falsification: drop the Bad-sample filter ⇒ the Uncertain interval test fails.
|
||||
|
||||
**Verify:** `dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests --filter "FullyQualifiedName~Read"` → green.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs","src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs"],"acceptanceCriteria":["raw fwd/rev","AtTime interp/step","aggregate table with hand-computed literals","unsupported aggregates throw"],"modelTier":"frontier","model":"opus","wave":3}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 11: SQL Server updater
|
||||
|
||||
**Goal:** `SqlServerHistorianDataUpdater : IHistorianDataUpdater` per spec §4 "Updates".
|
||||
|
||||
**Files (create only):** `.../SqlServerHistorianDataUpdater.cs`; tests `tests/.../SqlServer.IntegrationTests/UpdateTests.cs`.
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] One transaction per call; `MERGE Sample AS t USING @rows AS r ON t.TagId=@tagId AND t.SourceTimeUtc=r.SourceTimeUtc` with `OUTPUT $action, r.SourceTimeUtc`; Insert uses `WHEN NOT MATCHED THEN INSERT` only (matched ⇒ BadEntryExists), Replace `WHEN MATCHED THEN UPDATE` only (not matched ⇒ BadNoEntryExists), Update both. Results mapped back to the caller's original order (including within-request duplicate timestamps: second occurrence ⇒ `BadEntryExists` for Insert, last-wins for Replace/Update — document).
|
||||
- [ ] Null timestamp ⇒ `BadInvalidTimestamp` for that item, others proceed.
|
||||
- [ ] Delete raw `[start,end)` ⇒ `Good` with rows deleted, `BadNoData` when 0; DeleteAtTime per time.
|
||||
- [ ] Unknown tag: Insert/Update create it (via a private catalog lookup — do NOT edit T9's `SqlServerTagCatalog`; if T9 is not merged yet, implement a private `MERGE` in this file); Replace/Delete ⇒ BadNoEntryExists/BadNoData.
|
||||
- [ ] Integration test per row of the spec's result table + the duplicate/null cases; replaced value is visible via direct SQL with the new status.
|
||||
|
||||
**Verify:** `dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests --filter "FullyQualifiedName~Update"` → green.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataUpdater.cs"],"acceptanceCriteria":["Part 11 result codes","order preserved","single transaction"],"modelTier":"standard","model":"sonnet","wave":3}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 12: SQL Server provider + Host registration
|
||||
|
||||
**Goal:** `Backend=SqlServer` works end to end in the Host.
|
||||
|
||||
**Files:**
|
||||
- Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs` (one shared bootstrapper + tag catalog instance across the five seams; `SupportedKinds` = all six kinds)
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs` (add `"SqlServer"`, binding `ServerHistorian:SqlServer` to `SqlServerHistorianOptions` and failing on its `Validate()` errors), `src/Server/ZB.MOM.WW.OtOpcUa.Host/ZB.MOM.WW.OtOpcUa.Host.csproj` (ProjectReference), `Program.cs` only if T5 left a seam that needs it
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs` (extend), SqlServer.Tests `SqlServerHistorianProviderTests.cs`
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] `Backend=SqlServer` with no ConnectionString ⇒ host-start validation failure naming `ServerHistorian:SqlServer:ConnectionString`.
|
||||
- [ ] DI: `IHistorianDataSource` ⇒ `SqlServerHistorianDataSource`, `IHistorianValueWriter` ⇒ SqlServer writer, `HistorianUpdaterHolder.Updater` non-null, recorder `supportedKinds` contains String and Empty.
|
||||
- [ ] The Endpoint requirement does not fire for SqlServer.
|
||||
|
||||
**Verify:** `dotnet build ZB.MOM.WW.OtOpcUa.slnx && dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests --filter "FullyQualifiedName~Historian"` → green.
|
||||
|
||||
```json:metadata
|
||||
{"files":["src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs","src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs","src/Server/ZB.MOM.WW.OtOpcUa.Host/ZB.MOM.WW.OtOpcUa.Host.csproj"],"acceptanceCriteria":["validation names the key","DI resolves SqlServer types","updater present"],"modelTier":"standard","model":"sonnet","wave":4}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 13: Over-the-wire HistoryUpdate test
|
||||
|
||||
**Goal:** Prove HistoryUpdate works through a real SDK client session (the encoding/dispatch the node-manager unit tests mock around).
|
||||
|
||||
**Files (create only):** `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs` (use the existing integration harness that `ConditionMethodEventTests` uses; inject an in-memory fake `IHistorianDataUpdater` + `IHistorianDataSource` sharing one dictionary).
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] Client HistoryUpdate Insert of 2 values on a historized node ⇒ `GoodEntryInserted` ×2; HistoryRead raw returns them; Insert again ⇒ `BadEntryExists`; DeleteAtTime one ⇒ read returns one.
|
||||
- [ ] A session without the grant ⇒ `BadUserAccessDenied`, fake not called.
|
||||
- [ ] The node's AccessLevel read over the wire includes `HistoryWrite`.
|
||||
|
||||
**Verify:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests --filter "FullyQualifiedName~HistoryUpdateOverTheWire"` → green.
|
||||
|
||||
```json:metadata
|
||||
{"files":["tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs"],"acceptanceCriteria":["insert/read/duplicate/delete over the wire","deny case","HistoryWrite bit visible"],"modelTier":"standard","model":"sonnet","wave":3}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 14: Documentation
|
||||
|
||||
**Goal:** Docs match the shipped behaviour.
|
||||
|
||||
**Files:** `docs/Historian.md`, `docs/Configuration.md`, `docs/security.md`, `CLAUDE.md` (Historian / HistoryRead section: add backend selection, SQL Server, HistoryUpdate, the real-status fix, and the gateway C-002 quality limitation; keep it a short subsection), `docs/drivers/README` or index if historians are listed there.
|
||||
|
||||
**Acceptance Criteria:**
|
||||
- [ ] Every config key in spec §2 documented with default and validation rule.
|
||||
- [ ] Aggregate table + "simple per-sample definitions, not full Part 13" caveat.
|
||||
- [ ] HistoryUpdate: operations, result codes, authorization (flag on/off), AccessLevel bit, unsupported cases.
|
||||
- [ ] Docker-dev recipe for a SQL historian (`OtOpcUaHistorian` on the `sql` container) and the integration-test env var.
|
||||
- [ ] No claim the docs make is untested — cross-check each against the merged code.
|
||||
|
||||
**Verify:** controller reads the diff; `grep -n "Backend" docs/Configuration.md`.
|
||||
|
||||
```json:metadata
|
||||
{"files":["docs/Historian.md","docs/Configuration.md","docs/security.md","CLAUDE.md"],"acceptanceCriteria":["keys documented","aggregate caveat","HistoryUpdate auth documented"],"modelTier":"standard","model":"sonnet","wave":5}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Final review (fable), then Task 15: Live gate on docker-dev (controller)
|
||||
|
||||
**Final review:** dispatch a **fable** reviewer over `git diff master...feat/pluggable-historian` with the spec; it must specifically check #491 (`Self` after await), #537/#538 unchanged, every new status literal is a named const, ALLOW tests exist, no `Directory.Packages.props` edit, gateway Good-bytes unchanged. One fix dispatch (opus) + scoped re-review (sonnet), then adjudicate.
|
||||
|
||||
**Task 15 — Live gate (controller only; not delegated).** **USER-ORDERED GATE — NON-SKIPPABLE.** Close only after each item below is re-validated with captured output.
|
||||
|
||||
1. `CREATE DATABASE OtOpcUaHistorian` on `otopcua-dev-sql-1`; set on central-1 and central-2 (compose env): `ServerHistorian__Enabled=true`, `ServerHistorian__Backend=SqlServer`, `ServerHistorian__SqlServer__ConnectionString=Server=sql,1433;Database=OtOpcUaHistorian;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True`, `ContinuousHistorization__Enabled=true` + `OutboxPath`. Rebuild images (the rig image goes stale), `up -d`.
|
||||
2. Mark a rig tag historized (one numeric, one string) and deploy. Evidence: `hist.Sample` row count advances by exactly one per published value (Primary-only), status column real.
|
||||
3. `Client.CLI historyread` raw + processed (Average, 10 s) on the raw NodeId AND its UNS NodeId — same rows.
|
||||
4. `Client.CLI historyupdate` insert (new timestamp) ⇒ `GoodEntryInserted`; repeat ⇒ `BadEntryExists`; replace ⇒ `GoodEntryReplaced`; delete-at-time ⇒ `Good`; each confirmed by `historyread` and by direct SQL. Use an API key whose principal has a Cluster-scoped `HistoryUpdate` grant; a key with only WriteOperate ⇒ `BadUserAccessDenied`.
|
||||
5. Stop the Primary container; the Secondary resumes recording into the same table; no duplicate `(TagId, SourceTimeUtc)` (guaranteed by PK — also check count growth rate stays 1×); restart.
|
||||
6. Stop the driver's device (or blackhole) ⇒ a Bad sample stored with Bad status (SQL Server backend only).
|
||||
7. Record evidence in `docs/plans/2026-10-09-pluggable-historian-live-gate.md`; restore the rig config (Backend off) unless the user wants it kept.
|
||||
|
||||
```json:metadata
|
||||
{"files":["docs/plans/2026-10-09-pluggable-historian-live-gate.md"],"acceptanceCriteria":["Primary-only one row per value","historyread raw+processed via raw and UNS NodeIds","historyupdate insert/duplicate/replace/delete codes","WriteOperate-only key denied","failover continues recording","Bad sample stored Bad"],"userGate":true,"tags":["user-gate"],"verifyCommand":"Client.CLI historyread/historyupdate against opc.tcp://localhost:4840 + sqlcmd on otopcua-dev-sql-1","modelTier":"frontier","model":"controller","wave":5}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Finish
|
||||
|
||||
Merge `feat/pluggable-historian` → `master` only with the user's go-ahead; push only with approval. Then update the scadaproj `sister-project-history/otopcua.md` (on scadaproj `main`, pushed) and the index row only if the summary changed (driver/backends list). File Gitea follow-ups: QuestDB backend spec; retention/purge; Part 11 modified values; time-weighted aggregates.
|
||||
Reference in New Issue
Block a user