diff --git a/CLAUDE.md b/CLAUDE.md index 427e71b3..60d871cf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -822,7 +822,8 @@ only the recorder + outbox knobs: | `CommitIntervalMs` | `100` | Periodic-mode commit cadence; required `> 0` only under `Periodic` | | `DrainBatchSize` | `64` | Entries peeked + written per drain pass | | `DrainIntervalSeconds` | `2` | Steady drain cadence (and post-success reschedule) | -| `Capacity` | `0` | Max un-acked outbox entries before drop-oldest; `0` = unbounded | +| `Capacity` | `500000` | Max un-acked outbox entries before drop-oldest; `0` = unbounded. Non-zero **by default since #538** — the disk backstop for a historian refusing *every* write, the one case the per-tag quarantine deliberately cannot resolve | +| `MaxWriteAttempts` | `5` | #538: consecutive failed drain passes one tag may accumulate — counted **only across passes where other tags wrote successfully** — before its queued entries are abandoned so the FIFO head can advance | | `MinBackoffSeconds` | `1` | Initial retry backoff after a failed drain pass | | `MaxBackoffSeconds` | `30` | Cap on the exponential retry backoff | @@ -925,9 +926,9 @@ to know: the captured context, so the caller still resumes on the actor thread. The fake must complete on a **pool** thread (`await Task.Delay(1).ConfigureAwait(false)`), as a real per-entry fsync does. Only reverting the production fix exposed that the first regression test proved nothing. -- **Both pair nodes historize the same value** — the recorder has **no Primary gate**, unlike the alarm - store-and-forward sink. A redundant pair therefore sends duplicate live values per historized tag. - Deliberately recorded, not "fixed": gating it introduces a failover window with no capture. +- **Both pair nodes historized the same value** — the recorder had **no Primary gate**, unlike the alarm + store-and-forward sink, so a redundant pair sent duplicate live values per historized tag. Filed as + **#537** and **fixed** — see the paragraph below. **Gate COMPLETE — #491 CLOSED 2026-08-04.** Outbox append, drain, `WriteLiveValues`, **restart-convergence**, `EnsureTags` provisioning (`ensured=2 failed=0`) and **read-back** all ✅. The @@ -935,13 +936,34 @@ read-back leg ran against the licensed `wonder-sql-vd03` with the VPN up: two re *distinct non-zero* constants (`8224`, `120`) each read back as its own value — a `0`-valued source would have proved nothing (the #536 trap). -⚠️ **A second defect was found and FILED, not fixed (#538): one permanently-failing tag re-writes every -healthy tag on every drain pass, forever.** `DrainOnceAsync` writes **every** per-tag group and then acks -**all-or-nothing**, so one bad tag stops the FIFO head advancing while the good tags' values are -re-submitted to the historian on every retry — duplicate samples, and an outbox that never truncates -(`Capacity` defaults to `0` = unbounded; observed at 8.3 MB). The all-or-nothing ack is deliberate -(`RemoveAsync` truncates *through* an id, so partial acking could drop older unwritten entries) — the -consequence is what is undocumented. +**Both gate findings are now CLOSED — #537 (Primary gate) and #538 (poison tag). Two things to know +before touching either.** + +- **#538 — the discriminator is "did another tag write this pass?", NOT a failure count.** `DrainOnceAsync` + writes every per-tag group and acks **all-or-nothing** (deliberate: `RemoveAsync` truncates *through* an + id, so partial acking could drop older unwritten entries). One permanently-failing tag therefore pinned + the FIFO head while every *healthy* tag was re-submitted on every pass, forever — duplicate samples, an + outbox that never truncated (observed 8.3 MB; signature = the gateway's `ok` **and** `error` counters + climbing together). The drain now quarantines a tag after `MaxWriteAttempts` consecutive failures, but + ⚠️ **only counts a failure on a pass where some other tag succeeded** — otherwise an ordinary gateway + restart (where everything fails) would discard the entire backlog, converting a duplicate-samples bug + into a lost-history one. A count-only quarantine passes the poison-tag test and fails nothing else, so + the load-bearing test is the **falsification control** (`A_backend_that_fails_every_tag_abandons_nothing`). + Abandonment is also deferred until *every* failing tag is over budget, so nothing is discarded on a pass + that could not have advanced the head anyway. +- **#537 — it gates the CAPTURE, not the drain, and that is the opposite of the alarm sink.** The alarm + queue is in LocalDb and *replicates*, so the Secondary holds the Primary's rows and gating the drain + loses nothing. The historization outbox is a node-local FasterLog replicating to nobody: gating only its + drain leaves the standby appending forever to a queue no one empties (unbounded disk, then a flood of + stale duplicates on promotion). ⚠️ **And the peer that matters is one recording the SAME refs, not one + holding the same rows** — a node stands down only for a Primary in its own per-cluster mesh, and a node + with no cluster-scoped role never stands down at all. Under a fleet-wide election the elected Primary is + usually running another cluster's configuration, and deferring to it ends this cluster's historization + *silently* (the drain reports success on an empty outbox forever) — the same scope mismatch that once + suspended the alarm drain on every node of the rig at once. An unknown role captures. New verdict on + `RedundancyRoleView` behind **`IHistorizationRoleView`**, registered unconditionally on driver nodes + (NOT alongside the alarm `IRedundancyRoleView`, which only exists when that section is enabled — the + #511 registration accident). Also re-confirmed live: **`EnsureTags` provisioning fires only for newly-ADDED historized tags** — flipping an existing tag's `historianTagname` produces no provisioning at all diff --git a/docs/Historian.md b/docs/Historian.md index 89e0edb4..1d360b87 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -133,16 +133,7 @@ source it from there. > value. Fixed; pinned by a regression test whose outbox fake completes on a **pool** thread (a fake > returning `ValueTask.CompletedTask`, or even `await Task.Yield()`, cannot reproduce it). > -> ⚠️ **One permanently-failing tag re-writes every healthy tag, forever -> ([#538](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/538)).** `DrainOnceAsync` writes **every** -> per-tag group in a batch and then acks **all-or-nothing**, so a tag whose historian tag does not exist -> stops the FIFO head advancing while the healthy tags' values are re-submitted on every drain pass — -> duplicate samples at identical timestamps, and an outbox that never truncates (`Capacity` defaults to -> `0` = unbounded; observed at 8.3 MB). The all-or-nothing ack itself is deliberate. -> -> ⚠️ **Both nodes of a redundant pair historize the same value -> ([#537](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/537))** — the recorder has no Primary gate, -> unlike the alarm store-and-forward sink, so the historian receives duplicate live values per tag. +> The gate also found the two defects below, both since fixed. > > **Related, separately open:** `EnsureTags` provisioning fires only for **newly-added** historized > tags (`AddressSpaceApplier` iterates the added set), so flipping an **existing** tag to @@ -151,6 +142,71 @@ source it from there. --- +## A poison tag no longer pins the outbox (#538) + +`DrainOnceAsync` peeks a batch, writes **every** per-tag group in it, and then acks **all-or-nothing** — +because the outbox's `RemoveAsync` truncates the FIFO head *through* the acked id, so acking one tag's +entries could silently drop older, unwritten entries belonging to another. That much is deliberate and +unchanged. + +What was wrong is what happened when a tag failed **permanently**: the head never advanced, so every +**healthy** tag in the batch was re-submitted to the historian on every drain pass, indefinitely — +duplicate samples at identical timestamps, and an outbox that never truncated (observed live at 8.3 MB +from a single unprovisionable tag, with the gateway's `ok` **and** `error` counters climbing together — +the signature of this bug). + +The drain now tracks **consecutive failures per historian tag** and abandons a tag's queued entries once +it exceeds `ContinuousHistorization:MaxWriteAttempts` (default 5), letting the head advance. Two +properties matter: + +- **A pass where NOTHING wrote abandons nothing.** The counter only advances on a pass where at least one + *other* tag wrote successfully — that is what distinguishes "this tag is poison" from "the backend is + down". A count-only quarantine looks identical on the poison-tag case and silently discards the whole + backlog during an ordinary gateway restart, turning a duplicate-samples bug into a lost-history one. + Pinned by a falsification-control test (`A_backend_that_fails_every_tag_abandons_nothing`) that goes red + when the evidence check is removed while the poison-tag test stays green. +- **A tag over budget waits for the others.** Abandonment is deferred until *every* failing tag in the + batch is over budget, so a tag is never discarded on a pass that could not have advanced the head anyway. + +Each abandonment logs a Warning naming the tag and the discarded sample count, and the total rides +`RecorderStatus.AbandonedEntries`. The most common cause is a tag the historian has no definition for — +easy to reach, because `EnsureTags` provisioning runs only for newly-**added** historized tags, so a +changed `historianTagname` is never provisioned (Issue 5 above). + +`ContinuousHistorization:Capacity` now also defaults to **500,000** entries rather than unbounded. It is +the disk backstop for the one case the quarantine deliberately cannot resolve — a historian refusing +*every* write, where retaining is correct. Set `0` to restore unbounded retention. + +## Only the recording Primary captures (#537) + +Both nodes of a redundant pair serve the same address space and see the same dependency-mux fan-out, so +an ungated recorder wrote **every sample twice** (measured: the gateway's write counter advanced twice +per value with both centrals up, and halved when one was stopped). + +The recorder now reads a capture gate per value, fed from `PrimaryGatePolicy.ShouldRecordHistorizedValues` +via `IHistorizationRoleView` — the same shape as the alarm store-and-forward sink's `drainGate`, with two +deliberate differences: + +- **It gates the CAPTURE, not the drain.** The alarm queue lives in LocalDb and *replicates*, so the + Secondary already holds the Primary's rows and gating the drain loses nothing. The historization outbox + is a node-local FasterLog replicating to nobody: gating only its drain would leave the standby appending + forever to a queue no one empties — unbounded disk, and a flood of stale duplicates the moment it is + promoted. Entries a node already captured are still drained by that node, so a demotion loses nothing + recorded. +- **The relevant peer is one that records the same refs**, not one that holds the same rows. A node stands + down only for a Primary in its own per-cluster mesh. A node carrying no cluster-scoped role cannot + establish that, so it never stands down: under a fleet-wide election the elected Primary is usually + running a different cluster's configuration, and deferring to it would end this cluster's historization + silently — the drain would report success on an empty outbox forever. That is the same scope mismatch + that once suspended the alarm drain on every node of the docker-dev rig at once. + +An unknown role **captures**, matching the alarm drain: a duplicate sample is recoverable, a hole in the +trend is not. The residual cost is a capture gap across a failover, bounded by how long the newly promoted +node takes to see its first redundancy snapshot — inherent to any non-duplicating design. Suppressed +values are counted in `RecorderStatus.SuppressedNotPrimary`, and each gate transition logs once. + +--- + ## Tag auto-provisioning (EnsureTags) When a deploy adds a historized tag, the server auto-ensures it exists in the historian via the gateway's diff --git a/docs/Redundancy.md b/docs/Redundancy.md index 39550e53..e05d97a6 100644 --- a/docs/Redundancy.md +++ b/docs/Redundancy.md @@ -516,6 +516,20 @@ Under warm/hot redundancy both cluster nodes run `ScriptedAlarmHostActor` and ev - **`HistorianAdapterActor` historization** — likewise Primary-gated so alarm historization is exactly-once across all alarm sources. The actor subscribes to the `alerts` DPS topic and translates each `AlarmTransitionEvent` → `AlarmHistorianEvent` before enqueuing it on the sink; scripted alarms therefore historize exactly once regardless of cluster size. - **The alarm sink's DRAIN** — a second, independent gate on the same decision. The enqueue gate above is not sufficient any more: since Phase 2 the store-and-forward buffer lives in the replicated LocalDb, so the Secondary holds a full copy of the Primary's queued rows and an ungated drain there would re-deliver every event continuously. `LocalDbStoreAndForwardSink` takes a `drainGate` fed from `IRedundancyRoleView` — a singleton `DriverHostActor` publishes its `PrimaryGatePolicy` verdict to, because the drain runs on a timer and cannot receive `RedundancyStateChanged` itself. A gated-off node reports `HistorianDrainState.NotPrimary`. Delivery is **at-least-once across a failover** by design: rows the old Primary delivered just before dying may not have replicated their deletes yet. See [AlarmHistorian.md](AlarmHistorian.md). +- **Continuous-historization value CAPTURE** (Gitea #537) — a third gate on the same role rule, and the + one that gates the *opposite* half of its pipeline. Both pair nodes serve the same address space and see + the same dependency-mux fan-out, so an ungated `ContinuousHistorizationRecorder` wrote every live sample + twice. It now reads a capture gate per value, fed from `PrimaryGatePolicy.ShouldRecordHistorizedValues` + via `IHistorizationRoleView` (a third verdict on the same `RedundancyRoleView` object, registered + unconditionally on driver nodes rather than alongside the alarm section's `IRedundancyRoleView`). + Gating the **capture** rather than the drain is deliberate: unlike the alarm queue, the historization + outbox is a node-local FasterLog that replicates to nobody, so a drain-only gate would leave the standby + appending forever to a queue no one empties. Values a node already captured are still drained by that + node. ⚠️ **The peer it stands down for must record the same refs** — a Primary in its own per-cluster + mesh — not merely hold the same rows; a node with no cluster-scoped role never stands down, because a + fleet-wide election's Primary usually runs a different cluster's configuration and deferring to it would + end this cluster's historization silently. See [Historian.md](Historian.md). + Net effect: each alarm transition appears **once** on `/alerts` and would historize once, not once per node. See [ScriptedAlarms.md](ScriptedAlarms.md) and [AlarmTracking.md](AlarmTracking.md) for the scripted-alarm engine internals. diff --git a/docs/plans/2026-08-01-continuous-historization-live-gate.md b/docs/plans/2026-08-01-continuous-historization-live-gate.md index aa16f14f..ebeabbb2 100644 --- a/docs/plans/2026-08-01-continuous-historization-live-gate.md +++ b/docs/plans/2026-08-01-continuous-historization-live-gate.md @@ -248,8 +248,15 @@ because gating it would also mean a failover window with no capture at all. ## Follow-ups - **#491 CLOSED** 2026-08-04 — both legs proven. -- **#537** — the recorder has no Primary gate, so both pair nodes write duplicate live values. -- **#538** — one permanently-failing tag re-writes every healthy tag on every drain pass and the - outbox grows unbounded (the all-or-nothing ack above). +- **#537 CLOSED** — the recorder now gates its **capture** (not its drain — the outbox is node-local and + replicates to nobody, unlike the alarm queue) on `PrimaryGatePolicy.ShouldRecordHistorizedValues`, read + per value through `IHistorizationRoleView`. It stands down only for a Primary that records the same + refs — a partner in its own per-cluster mesh — and a node with no cluster-scoped role never stands down. + See `docs/Historian.md` §"Only the recording Primary captures". +- **#538 CLOSED** — the drain now abandons a tag's queued entries after `MaxWriteAttempts` consecutive + failures, counted **only across passes where some other tag wrote successfully**, so a whole-backend + outage still retains everything. The all-or-nothing ack is unchanged. `Capacity` also gained a non-zero + default (500,000) as the disk backstop for the total-outage case. See `docs/Historian.md` §"A poison tag + no longer pins the outbox". - Still open from before this gate: `EnsureTags` fires only for newly-*added* historized tags (`2026-06-27-historian-gateway-integration-issues.md` Issue 5) — re-confirmed live here. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 950909b7..255f2590 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -286,6 +286,22 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// private readonly IDriverPrimaryRoleSource? _primaryRoleSource; + /// + /// Gitea #537: where the continuous-historization capture verdict is published for the + /// recorder, which taps the mux fan-out and so needs the decision off its own mailbox. Null on + /// nodes that wire no redundancy manager. + /// + private readonly IHistorizationRoleView? _historizationRoleView; + + /// + /// Tests whether a node named in a redundancy snapshot serves this node's address space — i.e. + /// is a per-cluster mesh partner recording the very same historized refs. Null (the default) + /// means the caller could not establish it, which never stands this node down. See + /// for why proving the sharing + /// matters rather than trusting "somebody is Primary". + /// + private readonly Func? _sharesHistorizationScopeWithLocal; + /// /// Host of this node's LocalDb replication partner — the only node that holds a copy of this /// node's alarm queue, and therefore the only node it may ever stand down in favour of. @@ -422,7 +438,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers IDeploymentArtifactFetcher? artifactFetcher = null, IAlarmStateStore? alarmStateStore = null, ITelemetryLocalHub? telemetryHub = null, - IDriverPrimaryRoleSource? primaryRoleSource = null) => + IDriverPrimaryRoleSource? primaryRoleSource = null, + IHistorizationRoleView? historizationRoleView = null, + Func? sharesHistorizationScopeWithLocal = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -433,7 +451,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, - alarmStateStore, telemetryHub, primaryRoleSource)); + alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, + sharesHistorizationScopeWithLocal)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -480,6 +499,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// even though the production object implements both: a test that /// supplies only a role-view double must not silently also satisfy this, or its primary-role assertions /// pass without the publish ever happening. Null on nodes that wire no redundancy manager. + /// Gitea #537: where the continuous-historization capture + /// verdict is published, for the recorder that taps the mux fan-out. A third distinct parameter for + /// the same reason is a second — the production object + /// implements all three, and a test double supplied for one must not silently satisfy the others. + /// Null on nodes that wire no redundancy manager. + /// Tests whether a node named in a redundancy + /// snapshot serves this node's address space and therefore records the same historized refs. Null + /// (the default) means the caller could not establish it, and this node never stands down — see + /// . public DriverHostActor( IDbContextFactory? dbFactory, CommonsNodeId localNode, @@ -504,8 +532,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers IDeploymentArtifactFetcher? artifactFetcher = null, IAlarmStateStore? alarmStateStore = null, ITelemetryLocalHub? telemetryHub = null, - IDriverPrimaryRoleSource? primaryRoleSource = null) + IDriverPrimaryRoleSource? primaryRoleSource = null, + IHistorizationRoleView? historizationRoleView = null, + Func? sharesHistorizationScopeWithLocal = null) { + _historizationRoleView = historizationRoleView; + _sharesHistorizationScopeWithLocal = sharesHistorizationScopeWithLocal; _telemetryHub = telemetryHub; _deploymentArtifactCache = deploymentArtifactCache; _redundancyRoleView = redundancyRoleView; @@ -1431,6 +1463,22 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _redundancyRoleView?.Publish( PrimaryGatePolicy.ShouldDrainAlarmHistory(_localRole, queueSharingPeerIsPrimary: peerIsPrimary)); + // Gitea #537: the continuous-historization CAPTURE verdict. Same role rule as the drain gate + // above, but a different peer question — historization is not pair-local, it is fleet-wide by + // historian tag name, so the only node worth standing down for is one recording the SAME refs: + // a mesh partner serving this node's address space. With no way to establish that (a legacy + // node carrying no cluster-scoped role, or a test harness), the predicate is null and this + // node keeps capturing. Duplicate samples are recoverable; a hole in the trend is not. + var historizationPeerIsPrimary = + _sharesHistorizationScopeWithLocal is { } sharesScope + && msg.Nodes.Any(n => n.Role == RedundancyRole.Primary + && n.NodeId != _localNode + && sharesScope(n.NodeId)); + + _historizationRoleView?.PublishHistorizationRole( + PrimaryGatePolicy.ShouldRecordHistorizedValues( + _localRole, addressSpaceSharingPeerIsPrimary: historizationPeerIsPrimary)); + // The DATA-PLANE verdict, deliberately — the same one HandleRouteNodeWrite gates on, not the // drain verdict published immediately above. A driver reading this claims an exclusive identity // on a shared external system (Gitea #511: the Sparkplug host id), where two claimants is the diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/PrimaryGatePolicy.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/PrimaryGatePolicy.cs index 86a0e3bf..3ec410b2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/PrimaryGatePolicy.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/PrimaryGatePolicy.cs @@ -71,4 +71,56 @@ public static class PrimaryGatePolicy RedundancyRole.Secondary or RedundancyRole.Detached => !queueSharingPeerIsPrimary, _ => true, // unknown role ⇒ drain; a duplicate row beats a silent gap }; + + /// + /// Decide whether this node should capture continuous-historization values into its + /// durable outbox (Gitea #537). + /// + /// + /// + /// Same rule as , deliberately — and for the same + /// reason. The historian is append-only and the error costs are asymmetric: a false + /// allow costs a duplicate sample, a false deny costs a permanent hole in the trend. So an + /// unknown role captures, and only an explicit Secondary/Detached stands down. + /// + /// + /// But it gates the CAPTURE, not the drain — the opposite of the alarm sink. That + /// choice follows from where the queue lives. The alarm queue is in LocalDb and + /// replicates, so the Secondary already holds the Primary's rows and gating the + /// drain loses nothing. The historization outbox is a node-local FasterLog that replicates + /// to nobody: gating only its drain would leave the standby appending forever to a queue no + /// one empties (unbounded disk, and a flood of stale duplicates the moment it is promoted), + /// while gating the capture leaves it holding nothing. Values already captured are still + /// drained by the node that captured them, so a demotion loses nothing that was recorded. + /// + /// + /// The residual cost is a capture gap across a failover, bounded by how long the newly + /// promoted node takes to see its first snapshot. That is inherent to any non-duplicating + /// design; the alternative — both nodes capturing — is what #537 was filed about. + /// + /// + /// This node's last-known redundancy role, or null when unknown. + /// + /// Whether the node holding in the same snapshot is one + /// that serves this node's address space, and therefore captures the very same historized + /// refs — its per-cluster mesh partner. + /// + /// The scoping matters for the same reason it does on the alarm queue, and the mistake is + /// easier to make here because it looks safe: the redundancy election reaches only as far as + /// the caller can prove, whereas historization is fleet-wide by tag name. A node that stands + /// down in favour of a Primary running some other cluster's configuration hands its + /// tags to a node that never records them, and nothing says so — the drain simply reports + /// success on an empty outbox forever. Callers that cannot establish the sharing pass + /// false, which never stands down; duplicates are recoverable, a silent gap is not. + /// + /// + /// true to capture; false only when a node recording these same tags is doing it. + public static bool ShouldRecordHistorizedValues( + RedundancyRole? localRole, bool addressSpaceSharingPeerIsPrimary) => + localRole switch + { + RedundancyRole.Primary => true, + RedundancyRole.Secondary or RedundancyRole.Detached => !addressSpaceSharingPeerIsPrimary, + _ => true, // unknown role ⇒ capture; a duplicate sample beats a hole in the trend + }; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs index 74313a71..8e58c466 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs @@ -57,9 +57,33 @@ public sealed class ContinuousHistorizationOptions /// Steady drain cadence in seconds (also the post-success reschedule). Defaults to 2. public double DrainIntervalSeconds { get; init; } = 2; - /// Maximum un-acked outbox entries before the drop-oldest capacity policy kicks in; - /// 0 (the default) means unbounded. - public int Capacity { get; init; } + /// + /// Maximum un-acked outbox entries before the drop-oldest capacity policy kicks in; 0 + /// means unbounded. + /// + /// Defaults to 500,000 (Gitea #538) — roughly 50 MB of FasterLog, and hours-to-days of + /// backlog depending on tag count and update rate. It is a disk backstop for the one case + /// the per-tag quarantine cannot resolve: a historian that is unreachable or refusing + /// every write, where nothing proves any particular tag is at fault, so the + /// recorder correctly retains rather than discards. Set 0 to restore unbounded + /// retention where the outbox sits on storage that can absorb an arbitrarily long outage. + /// + /// + public int Capacity { get; init; } = 500_000; + + /// + /// Gitea #538: consecutive failed drain passes a single historian tag may accumulate — counted + /// only across passes where other tags wrote successfully, so a backend outage never + /// advances it — before that tag's queued entries are discarded and the FIFO head is allowed to + /// advance. Defaults to 5; clamped to at least 1 by the recorder. + /// + /// Without this, one permanently-failing tag (typically one whose historian tag was never + /// provisioned) pins the head forever and every healthy tag in the batch is + /// re-written to the historian on every pass, at cadence, + /// indefinitely. Each abandonment logs a Warning naming the tag. + /// + /// + public int MaxWriteAttempts { get; init; } = 5; /// Initial retry backoff (seconds) after a failed drain pass. Defaults to 1. public double MinBackoffSeconds { get; init; } = 1; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs index 1bd11017..b6adaf5c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs @@ -60,7 +60,12 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers } /// Outcome of one off-thread drain pass, piped back to the actor thread. - private sealed record DrainResult(bool Success, int Acked, int Attempted); + /// Whether the pass advanced the FIFO head (nothing left blocking it). + /// Entries durably written and truncated away. + /// Entries in the peeked batch. + /// Entries discarded because their tag is poison — see + /// (Gitea #538). Counted within . + private sealed record DrainResult(bool Success, int Acked, int Attempted, int Abandoned = 0); /// Ask the recorder for its current counters (used by the meter/health wiring). public sealed class GetStatus @@ -91,13 +96,19 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// Lifetime count of durable-boundary append failures (the value /// was dropped, not recorded, and the actor stayed alive rather than restart-looping). /// Whether the most recent drain pass acked cleanly. + /// Lifetime count of values not captured because this node is + /// the Secondary of a pair whose Primary records the same refs (Gitea #537). + /// Lifetime count of entries discarded because their historian tag + /// kept failing while other tags were writing successfully (Gitea #538). public sealed record RecorderStatus( int QueuedDepth, long TotalRecorded, long DroppedNonNumeric, long OutboxDropped, long OutboxAppendFailures, - bool LastDrainSucceeded); + bool LastDrainSucceeded, + long SuppressedNotPrimary = 0, + long AbandonedEntries = 0); private readonly IActorRef _dependencyMux; private readonly IHistorianValueWriter _writer; @@ -115,6 +126,20 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers private readonly TimeSpan _drainInterval; private readonly TimeSpan _minBackoff; private readonly TimeSpan _maxBackoff; + + /// Gitea #537: "should this node be capturing right now?". Supplied by the Runtime wiring + /// from the redundancy role view; a null gate (tests, single-node graphs) always captures. + private readonly Func _captureGate; + + /// Gitea #538: consecutive failed write passes a tag may accumulate — while OTHER tags are + /// writing successfully — before its queued entries are abandoned so the FIFO head can advance. + private readonly int _maxWriteAttempts; + + /// Consecutive failed write passes per historian tag, counted only across passes that + /// proved the backend reachable. Reset on the tag's first success. Mutated exclusively inside + /// , which keeps single-flight. + private readonly Dictionary _tagWriteFailures = new(StringComparer.Ordinal); + private readonly ILoggingAdapter _log = Context.GetLogger(); private readonly CancellationTokenSource _lifetime = new(); @@ -124,8 +149,14 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers private long _totalRecorded; private long _droppedNonNumeric; private long _outboxAppendFailures; + private long _suppressedNotPrimary; + private long _abandonedEntries; private bool _lastDrainSucceeded = true; + /// Last observed state of , so a change is logged once rather + /// than on every value. Seeded open: a gate that starts closed logs its first suppression. + private bool _captureGateOpen = true; + /// Gets or sets the timer scheduler (set by Akka via ). public ITimerScheduler Timers { get; set; } = null!; @@ -138,6 +169,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// Steady drain cadence; also the post-success reschedule. /// Initial retry backoff after a failed drain. /// Cap on the exponential retry backoff. + /// Gitea #537: "should this node be capturing right now?", read per + /// value. Null (the default) always captures — the correct posture for a single-node deployment + /// and for tests. + /// Gitea #538: consecutive failed write passes one tag may + /// accumulate, while other tags are succeeding, before its entries are abandoned. /// Props for the actor. public static Props Props( IActorRef dependencyMux, @@ -147,9 +183,12 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers int drainBatchSize = 64, TimeSpan? drainInterval = null, TimeSpan? minBackoff = null, - TimeSpan? maxBackoff = null) => + TimeSpan? maxBackoff = null, + Func? captureGate = null, + int maxWriteAttempts = 5) => Akka.Actor.Props.Create(() => new ContinuousHistorizationRecorder( - dependencyMux, writer, outbox, historizedRefs, drainBatchSize, drainInterval, minBackoff, maxBackoff)); + dependencyMux, writer, outbox, historizedRefs, drainBatchSize, drainInterval, minBackoff, maxBackoff, + captureGate, maxWriteAttempts)); /// Initializes a new instance of the class. /// The per-node dependency mux to register historized-ref interest with. @@ -160,6 +199,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// Steady drain cadence; also the post-success reschedule. /// Initial retry backoff after a failed drain. /// Cap on the exponential retry backoff. + /// Gitea #537: "should this node be capturing right now?", read per + /// value. Null (the default) always captures. + /// Gitea #538: consecutive failed write passes one tag may + /// accumulate, while other tags are succeeding, before its entries are abandoned. Clamped to at + /// least 1. public ContinuousHistorizationRecorder( IActorRef dependencyMux, IHistorianValueWriter writer, @@ -168,7 +212,9 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers int drainBatchSize = 64, TimeSpan? drainInterval = null, TimeSpan? minBackoff = null, - TimeSpan? maxBackoff = null) + TimeSpan? maxBackoff = null, + Func? captureGate = null, + int maxWriteAttempts = 5) { _dependencyMux = dependencyMux ?? throw new ArgumentNullException(nameof(dependencyMux)); _writer = writer ?? throw new ArgumentNullException(nameof(writer)); @@ -187,6 +233,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers _minBackoff = minBackoff is { } mb && mb > TimeSpan.Zero ? mb : TimeSpan.FromSeconds(1); _maxBackoff = maxBackoff is { } xb && xb > TimeSpan.Zero ? xb : TimeSpan.FromSeconds(30); _currentBackoff = _minBackoff; + // A null gate captures unconditionally. That is the safe default in both directions it is + // reached from: a deployment with no redundancy manager (nothing would ever open a closed + // gate), and a test that does not care about the role. + _captureGate = captureGate ?? (static () => true); + _maxWriteAttempts = maxWriteAttempts > 0 ? maxWriteAttempts : 1; ReceiveAsync(OnValueChangedAsync); Receive(OnUpdateHistorizedRefs); @@ -236,6 +287,33 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers return; } + // Gitea #537: both nodes of a redundant pair serve the same address space and see the same mux + // fan-out, so an ungated recorder writes every sample TWICE (measured on the docker-dev central + // pair: the gateway's write counter advanced twice per value, and halved when one node stopped). + // The gate closes only on a node whose pair partner is the Primary and is recording the same + // refs; it is read PER VALUE rather than latched, so promotion resumes capture on the very next + // sample with no restart. Note what is NOT gated: the drain. Entries already captured are this + // node's to deliver, and a demoted node must still empty its own outbox. + if (!_captureGate()) + { + _suppressedNotPrimary++; + if (_captureGateOpen) + { + _captureGateOpen = false; + _log.Info("ContinuousHistorization: this node is not the recording Primary for its address space; " + + "suppressing value capture until it is promoted (the pair Primary is recording these tags)."); + } + + return; + } + + if (!_captureGateOpen) + { + _captureGateOpen = true; + _log.Info("ContinuousHistorization: this node is now the recording Primary for its address space; " + + "resuming value capture."); + } + if (!TryCoerceToDouble(msg.Value, out double numeric)) { // The SQL analog live-value path is numeric-only — drop non-coercible values (string/null/ @@ -431,9 +509,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers } // Write per-tag batches. The outbox's RemoveAsync truncates the FIFO head THROUGH the acked - // id, so partial per-tag acking could drop un-written older entries; ack all-or-nothing — - // only when every tag's batch wrote do we truncate the whole peeked prefix (one commit). - var allOk = true; + // id, so partial per-tag acking could drop un-written older entries; the ack stays + // all-or-nothing — only when nothing is left blocking do we truncate the whole peeked + // prefix (one commit). + var succeeded = 0; + var failedGroups = new List<(string Tag, int Count)>(); foreach (IGrouping group in batch.GroupBy(e => e.Tag, StringComparer.Ordinal)) { @@ -453,21 +533,82 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers ok = false; } - if (!ok) + if (ok) { - allOk = false; + succeeded++; + _tagWriteFailures.Remove(group.Key); + } + else + { + failedGroups.Add((group.Key, values.Count)); } } - if (!allOk) + if (failedGroups.Count == 0) { - // Leave the entire peeked prefix queued for the next (backed-off) attempt. + // Ack the whole prefix by truncating through the last entry (FIFO head advance + commit). + await _outbox.RemoveAsync(batch[^1].Id, ct).ConfigureAwait(false); + return new DrainResult(true, batch.Count, batch.Count); + } + + if (succeeded == 0) + { + // NOTHING wrote this pass, so there is no evidence any particular tag is at fault — the + // backend itself is unreachable or refusing everything. Retain the whole prefix and back + // off; this is the outage path, and it must never discard data. Deliberately does not + // touch the per-tag counters either: they count consecutive failures observed while the + // backend was demonstrably up, so an outage neither advances nor resets them. return new DrainResult(false, 0, batch.Count); } - // Ack the whole prefix by truncating through the last entry (FIFO head advance + commit). + // Gitea #538: at least one tag wrote, so the backend IS reachable and a tag that keeps + // failing is poison — most often one whose historian tag was never provisioned (EnsureTags + // fires only for newly-ADDED historized tags, so a renamed historianTagname is easy to hit). + // Left blocking, it pins the FIFO head forever while every HEALTHY tag in the batch is + // re-written to the historian on every pass — duplicate samples, and an outbox that never + // truncates. Abandon the poison tag's entries once it has burned through its attempts, so + // the head can advance; the loss is bounded to one tag and it is loudly reported, whereas + // the alternative silently corrupts every other tag's history. + // Count first, decide second. A tag that is over its budget while ANOTHER is still within + // its own does not get abandoned yet — the head cannot advance either way, so abandoning it + // now would only discard entries early and log the same warning on every pass until the + // other tag resolves. + var stillBlocking = false; + foreach ((string tag, int _) in failedGroups) + { + int failures = _tagWriteFailures.GetValueOrDefault(tag) + 1; + _tagWriteFailures[tag] = failures; + if (failures <= _maxWriteAttempts) + { + stillBlocking = true; + } + } + + if (stillBlocking) + { + // Leave the entire peeked prefix queued for the next (backed-off) attempt. Bounded: + // every failing tag's counter advanced, so this can repeat at most _maxWriteAttempts + // times before the branch below takes over. + return new DrainResult(false, 0, batch.Count); + } + + var abandoned = 0; + foreach ((string tag, int count) in failedGroups) + { + abandoned += count; + _log.Warning( + "ContinuousHistorization: historian tag {Tag} has failed {Failures} consecutive write " + + "attempts while other tags wrote successfully; discarding {Count} queued sample(s) so " + + "the outbox can drain. Check the tag exists in the historian — EnsureTags provisioning " + + "runs only for newly-ADDED historized tags, so a changed historianTagname is never " + + "provisioned.", + tag, _tagWriteFailures.GetValueOrDefault(tag), count); + } + + // Every failure is now over budget; the head may advance. The abandoned entries are + // truncated away with the acked ones — that is what "abandoned" means here. await _outbox.RemoveAsync(batch[^1].Id, ct).ConfigureAwait(false); - return new DrainResult(true, batch.Count, batch.Count); + return new DrainResult(true, batch.Count - abandoned, batch.Count, abandoned); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -488,6 +629,9 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers { _draining = false; _lastDrainSucceeded = result.Success; + // Accumulated on the actor thread from the piped result rather than inside the off-thread drain, + // so the counter has a single writer and GetStatus never reads a torn value. + _abandonedEntries += result.Abandoned; if (result.Success) { @@ -515,7 +659,9 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers _droppedNonNumeric, _outbox.DroppedCount, _outboxAppendFailures, - _lastDrainSucceeded); + _lastDrainSucceeded, + _suppressedNotPrimary, + _abandonedEntries); } private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/IRedundancyRoleView.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/IRedundancyRoleView.cs index b450e9e2..9565c658 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/IRedundancyRoleView.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/IRedundancyRoleView.cs @@ -41,8 +41,43 @@ public interface IRedundancyRoleView } /// -/// Thread-safe + , each -/// backed by its own volatile field. +/// A singleton snapshot of the continuous-historization capture decision (Gitea #537), so +/// the recorder can ask "should this node be recording historized values right now?". +/// +/// +/// +/// A separate interface from even though one object +/// implements both, because the two are registered on different terms: +/// IRedundancyRoleView is registered by AddAlarmHistorian and so exists only +/// where the alarm section is enabled, while this one is registered unconditionally on every +/// driver-role node. Piggybacking on the alarm registration would silently drop the +/// historization gate on any pair that runs historization without alarm history — the same +/// registration accident Gitea #511 documents on the neighbouring verdict. +/// +/// +/// Publishes the decision rather than the role, so +/// stays the single place that +/// decides. Note it is not the alarm drain verdict: the two policies agree on the role +/// rule but disagree on what makes a peer relevant (a queue-sharing replication partner vs. a +/// node serving the same address space), and the wrong peer notion in either direction is a +/// silent gap. +/// +/// +public interface IHistorizationRoleView +{ + /// Whether this node should be capturing continuous-historization values right now. + bool ShouldRecordHistorizedValues { get; } + + /// Records a fresh decision. Called by DriverHostActor on every redundancy snapshot. + /// + /// The decision produced. + /// + void PublishHistorizationRole(bool shouldRecordHistorizedValues); +} + +/// +/// Thread-safe + + +/// , each backed by its own volatile field. /// /// /// @@ -55,7 +90,7 @@ public interface IRedundancyRoleView /// them is the bug that live gate already found once. /// /// -public sealed class RedundancyRoleView : IRedundancyRoleView, IDriverPrimaryRoleSource +public sealed class RedundancyRoleView : IRedundancyRoleView, IDriverPrimaryRoleSource, IHistorizationRoleView { // Seeded with the decision for "role unknown" rather than a bare false. An unpublished view and a // node whose role nothing ever reports are genuinely the same situation, and they must behave the @@ -71,9 +106,18 @@ public sealed class RedundancyRoleView : IRedundancyRoleView, IDriverPrimaryRole private volatile bool _shouldActAsPrimary = PrimaryGatePolicy.ShouldServiceAsPrimary(localRole: null, driverMemberCount: 1); + // Seeded for the unknown-role case from the same policy that will later publish here: a node + // nothing has reported a role for captures, exactly as an unpaired or single-node deployment must. + // Defaulting closed would leave a deployment that never publishes a role with no trend data at all. + private volatile bool _shouldRecordHistorizedValues = + PrimaryGatePolicy.ShouldRecordHistorizedValues(localRole: null, addressSpaceSharingPeerIsPrimary: false); + /// public bool ShouldDrainAlarmHistory => _shouldDrainAlarmHistory; + /// + public bool ShouldRecordHistorizedValues => _shouldRecordHistorizedValues; + /// public bool ShouldActAsPrimary => _shouldActAsPrimary; @@ -82,4 +126,8 @@ public sealed class RedundancyRoleView : IRedundancyRoleView, IDriverPrimaryRole /// public void PublishPrimaryRole(bool shouldActAsPrimary) => _shouldActAsPrimary = shouldActAsPrimary; + + /// + public void PublishHistorizationRole(bool shouldRecordHistorizedValues) + => _shouldRecordHistorizedValues = shouldRecordHistorizedValues; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 851a2b8c..eeda4156 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -92,6 +92,13 @@ public static class ServiceCollectionExtensions // publishes two verdicts into the same object rather than into two that disagree. services.TryAddSingleton(); services.TryAddSingleton(sp => sp.GetRequiredService()); + + // Gitea #537: the continuous-historization capture verdict, registered here — unconditionally on + // every driver-role node — for the same reason as the line above, and NOT alongside the alarm + // drain's IRedundancyRoleView. AddAlarmHistorian registers that one only when its own section is + // enabled, so piggybacking would leave a redundant pair that historizes values but not alarms + // with no gate at all, and both nodes would write every sample twice with nothing to say so. + services.TryAddSingleton(sp => sp.GetRequiredService()); return services; } @@ -436,7 +443,15 @@ public static class ServiceCollectionExtensions drainBatchSize: continuousOptions.DrainBatchSize, drainInterval: TimeSpan.FromSeconds(continuousOptions.DrainIntervalSeconds), minBackoff: TimeSpan.FromSeconds(continuousOptions.MinBackoffSeconds), - maxBackoff: TimeSpan.FromSeconds(continuousOptions.MaxBackoffSeconds)), + maxBackoff: TimeSpan.FromSeconds(continuousOptions.MaxBackoffSeconds), + // Gitea #537: read the verdict PER VALUE off the shared role view rather + // than capturing a bool, so a promotion resumes capture on the next sample. + // The view is registered unconditionally on driver nodes; a null one (test + // graphs) captures unconditionally, which is the right posture there. + captureGate: resolver.GetService() is { } histRoleView + ? () => histRoleView.ShouldRecordHistorizedValues + : null, + maxWriteAttempts: continuousOptions.MaxWriteAttempts), ContinuousHistorizationRecorderActorName); registry.Register(continuousRecorder); } @@ -494,6 +509,19 @@ public static class ServiceCollectionExtensions PeerProbeSupervisorName); registry.Register(peerProbes); + // Gitea #537: "does this snapshot node record the same historized refs as me?" — true only + // for a partner in this node's own per-cluster mesh, which is exactly the set that serves + // this node's address space. A node carrying NO cluster-scoped role (the legacy, + // pre-Phase-6 shape) cannot establish that, so the predicate stays null and this node never + // stands down: on a fleet-wide election the elected Primary is usually running a DIFFERENT + // cluster's configuration, and deferring to it would silently end this cluster's + // historization altogether — the same scope mismatch that once suspended the alarm drain on + // every node at once. + Func? historizationScopePeer = + roleInfo.ClusterRole is { } meshRole + ? nodeId => roleInfo.MembersWithRole(meshRole).Contains(nodeId) + : null; + var driverHost = system.ActorOf( // ackRouter: under ClusterClient mode the node comm actor relays ApplyAcks across // the boundary; under Dps it stays null and the host publishes on the @@ -528,7 +556,11 @@ public static class ServiceCollectionExtensions // Gitea #511: where this actor publishes its Primary DATA-PLANE verdict for drivers // that claim an exclusive external identity. Registered unconditionally by // AddOtOpcUaRuntime, unlike redundancyRoleView above — see the comment there. - primaryRoleSource: resolver.GetService()), + primaryRoleSource: resolver.GetService(), + // Gitea #537: where this actor publishes the continuous-historization CAPTURE + // verdict, and how it decides which peer is worth standing down for. + historizationRoleView: resolver.GetService(), + sharesHistorizationScopeWithLocal: historizationScopePeer), DriverHostActorName); registry.Register(driverHost); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistorizationRoleViewTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistorizationRoleViewTests.cs new file mode 100644 index 00000000..ad699c62 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistorizationRoleViewTests.cs @@ -0,0 +1,184 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Gitea #537 — the bridge that carries the continuous-historization capture verdict out of +/// the actor and into , where the recorder reads it per value. +/// +/// +/// +/// Both nodes of a redundant pair serve the same address space and see the same dependency-mux +/// fan-out, so an ungated recorder writes every sample twice. Measured on the docker-dev central +/// pair against a real gateway: the write counter advanced twice per value, and halved when one +/// node was stopped. +/// +/// +/// These cases share the role rule with but differ on +/// which peer counts. The alarm queue replicates, so the relevant peer is whoever holds a copy +/// of this node's rows; historization is fleet-wide by historian tag name, so the relevant peer +/// is whoever records this node's refs — a partner in its own per-cluster mesh. +/// +/// +public sealed class DriverHostActorHistorizationRoleViewTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-histview-test"); + private static readonly NodeId MeshPeer = NodeId.Parse("mesh-peer:4053"); + private static readonly NodeId ForeignNode = NodeId.Parse("other-cluster:4053"); + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); + + /// + /// Before any snapshot arrives the view must read open, not closed. + /// + /// + /// A deployment that runs no redundancy never publishes here at all. Defaulting closed would + /// leave it with no trend data whatsoever and nothing to say why. + /// + [Fact] + public void An_unpublished_view_captures() + => new RedundancyRoleView().ShouldRecordHistorizedValues.ShouldBeTrue(); + + /// A Secondary stands down for a Primary in its own mesh — the duplicate-write case. + [Fact] + public void Secondary_stands_down_for_a_primary_in_its_own_mesh() + { + var view = new RecordingHistorizationView(); + SpawnHost(view, sharesScopeWith: MeshPeer).Tell(Snapshot( + (TestNode, RedundancyRole.Secondary), (MeshPeer, RedundancyRole.Primary))); + + AwaitAssert(() => view.Published.ShouldBe([false]), duration: Timeout); + } + + /// + /// A Secondary whose Primary is in ANOTHER cluster keeps capturing. + /// + /// + /// That node runs a different configuration and records none of these tags, so standing down for + /// it would end this cluster's historization silently — the drain would simply report success on + /// an empty outbox forever. Same scope mismatch that once suspended the alarm drain fleet-wide. + /// + [Fact] + public void Secondary_keeps_capturing_when_the_primary_records_other_tags() + { + var view = new RecordingHistorizationView(); + SpawnHost(view, sharesScopeWith: MeshPeer).Tell(Snapshot( + (TestNode, RedundancyRole.Secondary), (ForeignNode, RedundancyRole.Primary))); + + AwaitAssert(() => view.Published.ShouldBe([true]), duration: Timeout); + } + + /// + /// With NO way to establish which peers share this node's address space, it never stands down. + /// + /// + /// The predicate is null on a node carrying no cluster-scoped role — the legacy, pre-Phase-6 + /// shape, where the redundancy election is fleet-wide and the elected Primary usually belongs to + /// some other cluster. Duplicates are recoverable; a silent gap is not, so unproven means + /// capture. Delete the null check in the wiring and this is the case that goes red. + /// + [Fact] + public void An_unprovable_scope_never_stands_down() + { + var view = new RecordingHistorizationView(); + SpawnHost(view, sharesScopeWith: null).Tell(Snapshot( + (TestNode, RedundancyRole.Secondary), (MeshPeer, RedundancyRole.Primary))); + + AwaitAssert(() => view.Published.ShouldBe([true]), duration: Timeout); + } + + /// Primary captures. + [Fact] + public void Primary_captures() + { + var view = new RecordingHistorizationView(); + SpawnHost(view, sharesScopeWith: MeshPeer).Tell(Snapshot( + (TestNode, RedundancyRole.Primary), (MeshPeer, RedundancyRole.Secondary))); + + // Published, not current — `true` is also the seed, so asserting the property would pass before + // the actor had processed anything. + AwaitAssert(() => view.Published.ShouldBe([true]), duration: Timeout); + } + + /// + /// Unknown role captures — the snapshot never names this node, the documented identity-mismatch + /// shape. + /// + [Fact] + public void Unknown_role_captures() + { + var view = new RecordingHistorizationView(); + SpawnHost(view, sharesScopeWith: MeshPeer).Tell(Snapshot( + (MeshPeer, RedundancyRole.Primary), (ForeignNode, RedundancyRole.Secondary))); + + AwaitAssert(() => view.Published.ShouldBe([true]), duration: Timeout); + } + + /// A promotion must MOVE the view, not merely leave it at its open seed. + [Fact] + public void A_promotion_reopens_a_closed_view() + { + var view = new RecordingHistorizationView(); + IActorRef host = SpawnHost(view, sharesScopeWith: MeshPeer); + + host.Tell(Snapshot((TestNode, RedundancyRole.Secondary), (MeshPeer, RedundancyRole.Primary))); + AwaitAssert(() => view.Published.ShouldBe([false]), duration: Timeout); + + host.Tell(Snapshot((TestNode, RedundancyRole.Primary), (MeshPeer, RedundancyRole.Secondary))); + AwaitAssert(() => view.Published.ShouldBe([false, true]), duration: Timeout); + } + + // ---------------- helpers ---------------- + + /// + /// A view that remembers every published decision, so a test asserts on the actor's ACTION + /// rather than on a state its open seed already satisfies. + /// + private sealed class RecordingHistorizationView : IHistorizationRoleView + { + private readonly List _published = []; + private readonly Lock _gate = new(); + private readonly RedundancyRoleView _inner = new(); + + /// Every value published so far, in order. + public IReadOnlyList Published + { + get { lock (_gate) return _published.ToArray(); } + } + + public bool ShouldRecordHistorizedValues => _inner.ShouldRecordHistorizedValues; + + public void PublishHistorizationRole(bool shouldRecordHistorizedValues) + { + lock (_gate) _published.Add(shouldRecordHistorizedValues); + _inner.PublishHistorizationRole(shouldRecordHistorizedValues); + } + } + + private IActorRef SpawnHost(IHistorizationRoleView view, NodeId? sharesScopeWith) => + Sys.ActorOf(DriverHostActor.Props( + NewInMemoryDbFactory(), + TestNode, + ackRouter: null, + driverMemberCountProvider: () => 2, + historizationRoleView: view, + sharesHistorizationScopeWithLocal: + sharesScopeWith is { } peer ? node => node == peer : null)); + + private static RedundancyStateChanged Snapshot(params (NodeId Node, RedundancyRole Role)[] nodes) => + new( + nodes.Select(n => new NodeRedundancyState( + n.Node, + n.Role, + IsClusterLeader: n.Role == RedundancyRole.Primary, + IsDriverPrimary: n.Role == RedundancyRole.Primary, + AsOfUtc: DateTime.UtcNow)) + .ToList(), + CorrelationId.NewId()); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/PrimaryGatePolicyTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/PrimaryGatePolicyTests.cs index 7a1262a9..24f71e09 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/PrimaryGatePolicyTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/PrimaryGatePolicyTests.cs @@ -119,3 +119,73 @@ public sealed class AlarmHistoryDrainGatePolicyTests PrimaryGatePolicy.ShouldDrainAlarmHistory(localRole: null, queueSharingPeerIsPrimary: true).ShouldBeTrue(); } } + +/// +/// Gitea #537 — the continuous-historization capture gate. +/// +/// +/// +/// Same role rule as the alarm drain, and for the same reason: the historian is append-only, so +/// a false allow costs a duplicate sample while a false deny costs a permanent hole in the +/// trend. What differs is the peer question — the alarm queue is pair-local and +/// replicated, so the peer that matters is whoever holds a copy of these rows; historization is +/// fleet-wide by tag name, so the peer that matters is whoever records these same refs. +/// +/// +/// It also gates the opposite half of the pipeline. The alarm sink gates its DRAIN, because the +/// Secondary already holds a replica of the Primary's queue. The historization outbox is +/// node-local and replicates to nobody, so gating only the drain would leave the standby +/// appending forever to a queue no one empties. See +/// PrimaryGatePolicy.ShouldRecordHistorizedValues. +/// +/// +public sealed class HistorizationCaptureGatePolicyTests +{ + [Theory] + [InlineData(RedundancyRole.Primary, false, true)] + [InlineData(RedundancyRole.Primary, true, true)] // Primary captures even if another claims it + [InlineData(RedundancyRole.Secondary, true, false)] // its mesh partner is recording these refs + [InlineData(RedundancyRole.Detached, true, false)] + public void Known_role_decides_against_the_peer( + RedundancyRole role, bool peerIsPrimary, bool expected) + => PrimaryGatePolicy.ShouldRecordHistorizedValues(role, peerIsPrimary).ShouldBe(expected); + + /// + /// A Secondary whose Primary does not record these refs keeps capturing. + /// + /// + /// This is the case that makes the peer argument load-bearing rather than decorative. Under a + /// fleet-wide election the elected Primary is usually running some other cluster's + /// configuration entirely, and it will never record this node's tags. A bare "somebody is + /// Primary ⇒ stand down" rule ends this cluster's historization outright, and reports nothing + /// wrong: the drain simply succeeds on an empty outbox forever. It is the same scope mismatch + /// that once suspended the alarm drain on every node of the docker-dev rig at once. + /// + [Fact] + public void A_secondary_whose_primary_records_other_tags_keeps_capturing() + => PrimaryGatePolicy + .ShouldRecordHistorizedValues(RedundancyRole.Secondary, addressSpaceSharingPeerIsPrimary: false) + .ShouldBeTrue("a Secondary must not defer to a Primary that records none of its tags"); + + /// + /// Unknown role captures — a deployment that never publishes a role must still historize. + /// + [Theory] + [InlineData(false)] + [InlineData(true)] + public void Unknown_role_captures_whatever_the_peer_says(bool peerIsPrimary) + => PrimaryGatePolicy.ShouldRecordHistorizedValues(null, peerIsPrimary).ShouldBeTrue(); + + /// + /// Pinned against a future "harmonisation": with no role known and a peer present, the + /// device-write gate denies while this one allows. + /// + [Fact] + public void It_diverges_from_the_write_gate_when_the_role_is_unknown() + { + PrimaryGatePolicy.ShouldServiceAsPrimary(localRole: null, driverMemberCount: 4).ShouldBeFalse(); + PrimaryGatePolicy + .ShouldRecordHistorizedValues(localRole: null, addressSpaceSharingPeerIsPrimary: true) + .ShouldBeTrue(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs index c0614e95..e9cd3145 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs @@ -386,6 +386,227 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart } + // ---------------- Gitea #538: one poison tag must not pin the FIFO head ---------------- + + /// + /// A tag that keeps failing while other tags write successfully is abandoned, so the head + /// advances and the HEALTHY tags stop being re-submitted (Gitea #538). + /// + /// + /// The bug: the ack is all-or-nothing over the peeked prefix, but every per-tag group is + /// written before that decision. So one permanently-failing tag left the head pinned while the + /// good tags' values were re-written to the historian on every drain pass, forever — duplicate + /// samples, and an outbox that never truncated (observed live at 8.3 MB). + /// + /// The load-bearing assertion is the BOUND on how many times the healthy tag was submitted, + /// not that the outbox eventually empties. Without the fix the outbox never empties, but a + /// test that only waits for empty would simply time out and look like a flake; asserting the + /// re-write count states the actual defect. + /// + /// + [Fact] + public async Task A_permanently_failing_tag_is_abandoned_so_healthy_tags_stop_being_rewritten() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = true }; + writer.SetFailing("Bad.Tag", true); + var outbox = new InMemoryOutbox(); + + const int maxAttempts = 2; + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "Good.Tag", "Bad.Tag" }, + minBackoff: TimeSpan.FromMilliseconds(20), + maxBackoff: TimeSpan.FromMilliseconds(50), + drainInterval: TimeSpan.FromMilliseconds(20), + maxWriteAttempts: maxAttempts)); + + rec.Tell(new VirtualTagActor.DependencyValueChanged("Good.Tag", 1.0, DateTime.UtcNow)); + rec.Tell(new VirtualTagActor.DependencyValueChanged("Bad.Tag", 2.0, DateTime.UtcNow)); + + // The head advances: both entries leave the outbox — the good one written, the bad one discarded. + await AwaitAssertAsync(async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(0, status.QueuedDepth); + Assert.Equal(1, status.AbandonedEntries); + }, TimeSpan.FromSeconds(10)); + + // And the healthy tag was submitted a BOUNDED number of times — once per blocked pass plus the + // pass that unblocked, never indefinitely. Snapshot it, let several more drain intervals elapse, + // and prove it has stopped growing: an unbounded re-write would still be climbing. + int rewrites = writer.Snapshot().Count(w => w.Tag == "Good.Tag"); + Assert.InRange(rewrites, 1, maxAttempts + 1); + await Task.Delay(TimeSpan.FromMilliseconds(300)); + Assert.Equal(rewrites, writer.Snapshot().Count(w => w.Tag == "Good.Tag")); + } + + /// + /// When NOTHING writes, nothing is abandoned — the outage path must still retain everything. + /// + /// + /// This is the falsification control for the test above, and it is the reason the fix keys on + /// "another tag succeeded this pass" rather than on a bare per-tag failure count. A count-only + /// quarantine looks identical on the poison-tag case and silently DISCARDS the whole backlog + /// during an ordinary gateway restart — turning a bug about duplicate samples into one about + /// lost history. Delete the succeeded == 0 guard and this goes red while the other stays + /// green. + /// + [Fact] + public async Task A_backend_that_fails_every_tag_abandons_nothing() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = false }; // the whole backend is down + var outbox = new InMemoryOutbox(); + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B" }, + minBackoff: TimeSpan.FromMilliseconds(20), + maxBackoff: TimeSpan.FromMilliseconds(50), + drainInterval: TimeSpan.FromMilliseconds(20), + maxWriteAttempts: 1)); // the tightest budget there is — still must not discard + + rec.Tell(new VirtualTagActor.DependencyValueChanged("A", 1.0, DateTime.UtcNow)); + rec.Tell(new VirtualTagActor.DependencyValueChanged("B", 2.0, DateTime.UtcNow)); + + // Wait until the drain has retried well past the attempt budget, so "nothing abandoned" is a + // statement about a recorder that had every chance to abandon — not about one that hasn't run. + await AwaitAssertAsync( + () => + { + Assert.True(writer.CallCount >= 6, "the drain must have retried several times"); + return Task.CompletedTask; + }, + TimeSpan.FromSeconds(10)); + + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(0, status.AbandonedEntries); + Assert.Equal(2, status.QueuedDepth); // both values still durably queued for when it returns + } + + /// A tag that recovers inside its budget is never abandoned, and its counter resets. + [Fact] + public async Task A_tag_that_recovers_within_its_budget_is_not_abandoned() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = true }; + writer.SetFailing("Flaky.Tag", true); + var outbox = new InMemoryOutbox(); + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "Good.Tag", "Flaky.Tag" }, + minBackoff: TimeSpan.FromMilliseconds(20), + maxBackoff: TimeSpan.FromMilliseconds(50), + drainInterval: TimeSpan.FromMilliseconds(20), + maxWriteAttempts: 50)); // a budget the test will not exhaust + + rec.Tell(new VirtualTagActor.DependencyValueChanged("Good.Tag", 1.0, DateTime.UtcNow)); + rec.Tell(new VirtualTagActor.DependencyValueChanged("Flaky.Tag", 2.0, DateTime.UtcNow)); + + // It has genuinely failed a few times before recovering — otherwise this proves nothing. + await AwaitAssertAsync( + () => + { + Assert.True(writer.Snapshot().Count(w => w.Tag == "Flaky.Tag") >= 2); + return Task.CompletedTask; + }, + TimeSpan.FromSeconds(10)); + + writer.SetFailing("Flaky.Tag", false); + + await AwaitAssertAsync(async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(0, status.QueuedDepth); + Assert.Equal(0, status.AbandonedEntries); // recovered, so nothing was ever discarded + }, TimeSpan.FromSeconds(10)); + } + + // ---------------- Gitea #537: only the recording Primary captures ---------------- + + /// + /// A closed capture gate drops the value instead of appending it (Gitea #537). + /// + /// + /// Both nodes of a redundant pair serve the same address space and see the same mux fan-out, so + /// an ungated recorder writes every sample twice. + /// + /// The load-bearing assertion is SuppressedNotPrimary == 1, a POSITIVE observation + /// that the value arrived and was refused. "The outbox is empty" and "the writer saw + /// nothing" are both true before the recorder has processed anything at all, so a test built + /// on them alone passes against a recorder that never ran. + /// + /// + [Fact] + public async Task A_closed_capture_gate_suppresses_the_value() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter(); + var outbox = new InMemoryOutbox(); + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "Pump1.Temp" }, + captureGate: () => false)); + + rec.Tell(new VirtualTagActor.DependencyValueChanged("Pump1.Temp", 42.0, DateTime.UtcNow)); + + await AwaitAssertAsync(async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(1, status.SuppressedNotPrimary); + Assert.Equal(0, status.TotalRecorded); + }); + Assert.Empty(writer.Snapshot()); + } + + /// + /// The gate is read per value, so a promotion resumes capture on the very next sample. + /// + /// + /// Latching the verdict at construction (or on the first value) would compile, pass the + /// closed-gate test above, and leave a newly promoted node recording nothing until it restarted + /// — a silent, permanent gap that only a failover reveals. Flipping the gate mid-test is what + /// distinguishes the two implementations. + /// + [Fact] + public async Task The_capture_gate_is_re_read_per_value_so_promotion_resumes_capture() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = true }; + var outbox = new InMemoryOutbox(); + var open = false; + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "Pump1.Temp" }, + drainInterval: TimeSpan.FromMilliseconds(20), + captureGate: () => Volatile.Read(ref open))); + + rec.Tell(new VirtualTagActor.DependencyValueChanged("Pump1.Temp", 11.0, DateTime.UtcNow)); + await AwaitAssertAsync(async () => + { + var suppressed = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(1, suppressed.SuppressedNotPrimary); + }); + + Volatile.Write(ref open, true); // promoted + rec.Tell(new VirtualTagActor.DependencyValueChanged("Pump1.Temp", 22.0, DateTime.UtcNow)); + + await AwaitAssertAsync(() => + { + // Only the post-promotion value: the suppressed one was never appended, so it cannot appear + // late. Asserting on the CONTENT rather than the count is what rules out a gate that simply + // buffered the first value and released it. + var written = writer.Snapshot(); + Assert.Contains(written, w => w.Value == 22.0); + Assert.DoesNotContain(written, w => w.Value == 11.0); + return Task.CompletedTask; + }, TimeSpan.FromSeconds(10)); + } + /// In-memory double: records every value written and /// returns as the ack. Thread-safe — the recorder drains off the actor thread. private sealed class FakeValueWriter : IHistorianValueWriter @@ -401,6 +622,25 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit /// leaving the plain behaviour (FailFirstN==0) untouched. public int FailFirstN { get; init; } + /// Historian tags whose writes always fail, whatever says — the + /// poison-tag shape Gitea #538 is about (typically a tag the historian has no definition for, + /// because EnsureTags only provisions newly-ADDED historized tags). Other tags in the same drain + /// pass still succeed, and that contrast is the evidence the recorder keys on. + private readonly HashSet _failTags = new(StringComparer.Ordinal); + + /// Starts or stops failing one tag. Callable mid-test so a case can prove a tag that + /// recovers within its budget is never abandoned. + /// The historian tag. + /// Whether writes for it should fail. + public void SetFailing(string tag, bool failing) + { + lock (_gate) + { + if (failing) _failTags.Add(tag); + else _failTags.Remove(tag); + } + } + /// Lifetime count of invocations (proves a retry ran). public int CallCount { @@ -432,8 +672,9 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit _written.Add(new WrittenValue(tag, v.Value, v.Quality, v.TimestampUtc)); } - // Fail the first FailFirstN calls; thereafter honour Succeed. - bool result = _calls > FailFirstN && Succeed; + // Fail the first FailFirstN calls; thereafter honour Succeed — unless this specific tag + // is currently marked failing, which overrides both. + bool result = _calls > FailFirstN && Succeed && !_failTags.Contains(tag); return Task.FromResult(result); } }