Both defects were found by #491's live gate against a real HistorianGateway and filed rather than fixed at the time, because each one's fix is a design decision rather than a patch. #537 — both nodes of a redundant pair captured every value, so the historian received duplicate live samples for every historized tag (measured: the gateway's write counter advanced twice per value and halved when one central was stopped). The recorder now reads a capture gate per value, fed from the new PrimaryGatePolicy.ShouldRecordHistorizedValues through IHistorizationRoleView — a third verdict on the same RedundancyRoleView object, registered unconditionally on driver nodes rather than alongside the alarm section's IRedundancyRoleView (which only exists when that section is enabled; piggybacking is the #511 registration accident). Two things are deliberate and easy to get backwards: * It gates the CAPTURE, not the drain — the opposite of the alarm sink. That queue is in LocalDb and replicates, so the Secondary already holds the Primary's rows. The historization outbox is a node-local FasterLog replicating to nobody, so a drain-only gate would leave the standby appending forever to a queue no one empties: unbounded disk, then a flood of stale duplicates the moment it is promoted. Entries a node already captured are still drained by that node. * The peer it stands down for must RECORD THE SAME REFS, not merely hold the same rows — a Primary in this node's own per-cluster mesh. A node carrying no cluster-scoped role cannot establish that and 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, with the drain reporting success on an empty outbox forever. That is the scope mismatch that once suspended the alarm drain on every node of the rig. An unknown role captures, matching the alarm drain — a duplicate sample is recoverable, a hole in the trend is not. #538 — the drain writes every per-tag group and then 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 in the batch was re-submitted to the historian on every pass, indefinitely — duplicate samples and an outbox that never truncated, observed live at 8.3 MB from a single unprovisionable tag. The drain now tracks consecutive failures per historian tag and abandons that tag's entries once it exceeds MaxWriteAttempts (new, default 5), so the head advances. The counter only advances on a pass where some OTHER tag wrote successfully: that is what separates "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. 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. Each abandonment logs a Warning naming the tag; the total rides RecorderStatus.AbandonedEntries. Capacity also gains a non-zero default (500,000 entries) as 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. Every new test was falsified by reverting the production change it guards, and each one reddens alone: * drop the backend-reachable evidence -> only A_backend_that_fails_every_tag_abandons_nothing * never abandon -> only A_permanently_failing_tag_is_abandoned_so_healthy_tags_stop_being_rewritten * latch the capture gate at construction -> only The_capture_gate_is_re_read_per_value... * treat any Primary as a scope peer -> only the two scope-mismatch cases Full suite green. (Four OpcUaClient.Browser failures during the run were the opc-plc fixture being down on the shared Docker host; brought it back up and they pass.) Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
This commit is contained in:
Joseph Doherty
+184
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #537 — the bridge that carries the continuous-historization <b>capture</b> verdict out of
|
||||
/// the actor and into <see cref="IHistorizationRoleView"/>, where the recorder reads it per value.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// These cases share the role rule with <see cref="DriverHostActorRoleViewTests"/> 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
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);
|
||||
|
||||
/// <summary>
|
||||
/// Before any snapshot arrives the view must read open, not closed.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public void An_unpublished_view_captures()
|
||||
=> new RedundancyRoleView().ShouldRecordHistorizedValues.ShouldBeTrue();
|
||||
|
||||
/// <summary>A Secondary stands down for a Primary in its own mesh — the duplicate-write case.</summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A Secondary whose Primary is in ANOTHER cluster keeps capturing.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// With NO way to establish which peers share this node's address space, it never stands down.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>Primary captures.</summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Unknown role captures — the snapshot never names this node, the documented identity-mismatch
|
||||
/// shape.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>A promotion must MOVE the view, not merely leave it at its open seed.</summary>
|
||||
[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 ----------------
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private sealed class RecordingHistorizationView : IHistorizationRoleView
|
||||
{
|
||||
private readonly List<bool> _published = [];
|
||||
private readonly Lock _gate = new();
|
||||
private readonly RedundancyRoleView _inner = new();
|
||||
|
||||
/// <summary>Every value published so far, in order.</summary>
|
||||
public IReadOnlyList<bool> 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());
|
||||
}
|
||||
@@ -119,3 +119,73 @@ public sealed class AlarmHistoryDrainGatePolicyTests
|
||||
PrimaryGatePolicy.ShouldDrainAlarmHistory(localRole: null, queueSharingPeerIsPrimary: true).ShouldBeTrue();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #537 — the continuous-historization <b>capture</b> gate.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// 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 <i>peer</i> 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// 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
|
||||
/// <c>PrimaryGatePolicy.ShouldRecordHistorizedValues</c>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
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);
|
||||
|
||||
/// <summary>
|
||||
/// A Secondary whose Primary does not record these refs keeps capturing.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
[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");
|
||||
|
||||
/// <summary>
|
||||
/// Unknown role captures — a deployment that never publishes a role must still historize.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public void Unknown_role_captures_whatever_the_peer_says(bool peerIsPrimary)
|
||||
=> PrimaryGatePolicy.ShouldRecordHistorizedValues(null, peerIsPrimary).ShouldBeTrue();
|
||||
|
||||
/// <summary>
|
||||
/// Pinned against a future "harmonisation": with no role known and a peer present, the
|
||||
/// device-write gate denies while this one allows.
|
||||
/// </summary>
|
||||
[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();
|
||||
}
|
||||
}
|
||||
|
||||
+243
-2
@@ -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 ----------------
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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).
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[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.RecorderStatus>(
|
||||
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"));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// When NOTHING writes, nothing is abandoned — the outage path must still retain everything.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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 <c>succeeded == 0</c> guard and this goes red while the other stays
|
||||
/// green.
|
||||
/// </remarks>
|
||||
[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.RecorderStatus>(
|
||||
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
|
||||
}
|
||||
|
||||
/// <summary>A tag that recovers inside its budget is never abandoned, and its counter resets.</summary>
|
||||
[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.RecorderStatus>(
|
||||
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 ----------------
|
||||
|
||||
/// <summary>
|
||||
/// A closed capture gate drops the value instead of appending it (Gitea #537).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// <para>
|
||||
/// The load-bearing assertion is <c>SuppressedNotPrimary == 1</c>, 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[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.RecorderStatus>(
|
||||
ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
|
||||
Assert.Equal(1, status.SuppressedNotPrimary);
|
||||
Assert.Equal(0, status.TotalRecorded);
|
||||
});
|
||||
Assert.Empty(writer.Snapshot());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The gate is read per value, so a promotion resumes capture on the very next sample.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
[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.RecorderStatus>(
|
||||
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));
|
||||
}
|
||||
|
||||
/// <summary>In-memory <see cref="IHistorianValueWriter"/> double: records every value written and
|
||||
/// returns <see cref="Succeed"/> as the ack. Thread-safe — the recorder drains off the actor thread.</summary>
|
||||
private sealed class FakeValueWriter : IHistorianValueWriter
|
||||
@@ -401,6 +622,25 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
|
||||
/// leaving the plain <see cref="Succeed"/> behaviour (FailFirstN==0) untouched.</summary>
|
||||
public int FailFirstN { get; init; }
|
||||
|
||||
/// <summary>Historian tags whose writes always fail, whatever <see cref="Succeed"/> 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.</summary>
|
||||
private readonly HashSet<string> _failTags = new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>Starts or stops failing one tag. Callable mid-test so a case can prove a tag that
|
||||
/// recovers within its budget is never abandoned.</summary>
|
||||
/// <param name="tag">The historian tag.</param>
|
||||
/// <param name="failing">Whether writes for it should fail.</param>
|
||||
public void SetFailing(string tag, bool failing)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (failing) _failTags.Add(tag);
|
||||
else _failTags.Remove(tag);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Lifetime count of <see cref="WriteLiveValuesAsync"/> invocations (proves a retry ran).</summary>
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user