dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge commit '82380d2b' into remed/integration-w3

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-22 17:21:38 -04:00
parent 5533de6281 82380d2bee
commit 8699614d2f
60 changed files with 5954 additions and 752 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs
+176
View File
@@ -0,0 +1,176 @@
using System.Collections.Immutable;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb;
/// <summary>
/// Review R05:4 — the scripted-alarm comment trail moved off the condition row into its own
/// append-only <c>alarm_condition_comment</c> table. This proves the move kept the property that
/// mattered: the trail still reaches the redundant pair peer, over a real loopback h2c transport
/// through the real fail-closed interceptor, with both nodes initialised by the production
/// <c>LocalDbSetup.OnReady</c>.
/// </summary>
/// <remarks>
/// State is written and read through the production <see cref="LocalDbAlarmConditionStateStore"/>
/// on each node, so the ordinal keying, the incremental append and the load-side merge under
/// test are the ones production runs.
/// </remarks>
[Collection("LocalDbPairConvergence")]
public sealed class AlarmConditionCommentConvergenceTests
{
private static readonly DateTime T0 = new(2026, 06, 10, 12, 00, 00, DateTimeKind.Utc);
private static LocalDbAlarmConditionStateStore Store(ILocalDb db) =>
new(db, NullLogger<LocalDbAlarmConditionStateStore>.Instance);
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static async Task<int> CommentCountAsync(ILocalDb db, string alarmId) =>
(await Store(db).LoadAsync(alarmId, Ct))?.Comments.Count ?? -1;
[Fact]
public async Task CommentsWrittenOnA_ArriveOnB_AndBsAppendArrivesBackOnA()
{
await using var pair = new LocalDbPairHarness();
await pair.StartAsync();
var storeA = Store(pair.A);
var state = AlarmConditionState.Fresh("eq/alarm-1", T0) with
{
Active = AlarmActiveState.Active,
Acked = AlarmAckedState.Unacknowledged,
};
state = Part9StateMachine.ApplyAddComment(state, "jane", "first look", T0.AddMinutes(1)).State;
state = Part9StateMachine.ApplyAcknowledge(state, "bob", "on it", T0.AddMinutes(2)).State;
await storeA.SaveAsync(state, Ct);
await LocalDbPairHarness.WaitUntilAsync(
async () => await CommentCountAsync(pair.B, "eq/alarm-1") == 2,
"node B to hold node A's two comment-trail entries");
var onB = await Store(pair.B).LoadAsync("eq/alarm-1", Ct);
onB.ShouldNotBeNull();
onB.Comments.ShouldBe(state.Comments);
onB.Acked.ShouldBe(AlarmAckedState.Acknowledged);
// The reverse direction, as an incremental append: B adds one entry to the trail it loaded,
// and only that entry has to travel for A to see the full trail.
var appended = Part9StateMachine.ApplyAddComment(onB, "amy", "from B", T0.AddMinutes(3)).State;
await Store(pair.B).SaveAsync(appended, Ct);
await LocalDbPairHarness.WaitUntilAsync(
async () => await CommentCountAsync(pair.A, "eq/alarm-1") == 3,
"node A to receive the entry node B appended");
(await storeA.LoadAsync("eq/alarm-1", Ct))!.Comments.ShouldBe(appended.Comments);
}
[Fact]
public async Task Failover_AfterAMissedCommand_BothTheMissedEntryAndTheSurvivorsNextEntry_ReachBothPeers()
{
// Review I-1, end to end over the real pair:
// 1. a shared trail on both nodes;
// 2. an AddComment reaches only node A (B's command delivery failed) — A's entry replicates
// into B's table, but B's engine never applied it, so B's IN-MEMORY trail is one short;
// 3. A goes away (link down) and the operator acks on the survivor B — B's in-memory Ack
// collides with A's entry at the same ordinal;
// 4. the pair rejoins.
// Both entries must survive on both peers. Before the fix B's Ack was silently dropped.
await using var pair = new LocalDbPairHarness();
await pair.StartAsync();
var shared = AlarmConditionState.Fresh("eq/alarm-fo", T0) with
{
Active = AlarmActiveState.Active,
Acked = AlarmAckedState.Unacknowledged,
};
for (var i = 0; i < 3; i++)
shared = Part9StateMachine.ApplyAddComment(shared, "jane", $"note {i}", T0.AddMinutes(i)).State;
await Store(pair.A).SaveAsync(shared, Ct);
await LocalDbPairHarness.WaitUntilAsync(
async () => await CommentCountAsync(pair.B, "eq/alarm-fo") == 3,
"node B to hold the shared trail");
var onlyOnA = Part9StateMachine.ApplyAddComment(shared, "jane", "seen on A only", T0.AddHours(1)).State;
await Store(pair.A).SaveAsync(onlyOnA, Ct);
await LocalDbPairHarness.WaitUntilAsync(
async () => await CommentCountAsync(pair.B, "eq/alarm-fo") == 4,
"node A's entry to replicate into node B's table");
await pair.StopPassiveAsync();
// B's engine still holds the 3-entry trail it applied; the operator acks there.
var ackOnB = Part9StateMachine.ApplyAcknowledge(shared, "bob", "acked on B", T0.AddHours(2)).State;
await Store(pair.B).SaveAsync(ackOnB, Ct);
await pair.RestartPairAsync();
await LocalDbPairHarness.WaitUntilAsync(
async () => await CommentCountAsync(pair.A, "eq/alarm-fo") == 5
&& await CommentCountAsync(pair.B, "eq/alarm-fo") == 5,
"both nodes to hold the shared trail + A's missed-by-B entry + B's ack");
var onA = (await Store(pair.A).LoadAsync("eq/alarm-fo", Ct))!;
var onB = (await Store(pair.B).LoadAsync("eq/alarm-fo", Ct))!;
onA.Comments.ShouldBe(onB.Comments);
onA.Comments[3].ShouldBe(onlyOnA.Comments[3]);
onA.Comments[4].ShouldBe(ackOnB.Comments[3]);
// Held, not merely reached: nothing late duplicates either entry.
(await LocalDbPairHarness.HeldWithinAsync(
async () => await CommentCountAsync(pair.A, "eq/alarm-fo") != 5
|| await CommentCountAsync(pair.B, "eq/alarm-fo") != 5,
TimeSpan.FromSeconds(2)))
.ShouldBeFalse();
}
[Fact]
public async Task BothNodesRecordingTheSameEntry_ConvergeToOneEntry_NotTwo()
{
// Both pair nodes run the engine and both apply every operator command, each stamping the
// entry with its own clock. Ordinal keying is what collapses the two copies to one entry;
// a content-derived key would leave the trail doubled on both nodes.
await using var pair = new LocalDbPairHarness();
await pair.StartAsync();
var fresh = AlarmConditionState.Fresh("eq/alarm-2", T0);
var onA = Part9StateMachine.ApplyAddComment(fresh, "jane", "checked", T0.AddSeconds(1)).State;
var onB = Part9StateMachine.ApplyAddComment(fresh, "jane", "checked", T0.AddSeconds(1).AddMilliseconds(7)).State;
await Store(pair.A).SaveAsync(onA, Ct);
await Store(pair.B).SaveAsync(onB, Ct);
// Positive control: a DISTINCT second alarm written only on B. Without it "one entry on
// both" would also hold for two databases that never replicated at all.
var onlyOnB = Part9StateMachine.ApplyAddComment(
AlarmConditionState.Fresh("eq/only-on-b", T0), "amy", "b only", T0.AddSeconds(2)).State;
await Store(pair.B).SaveAsync(onlyOnB, Ct);
await LocalDbPairHarness.WaitUntilAsync(
async () => await CommentCountAsync(pair.A, "eq/only-on-b") == 1,
"node A to receive node B's distinct alarm");
// LWW picks ONE node's copy and both nodes end up holding that same copy.
await LocalDbPairHarness.WaitUntilAsync(
async () =>
{
var a = await Store(pair.A).LoadAsync("eq/alarm-2", Ct);
var b = await Store(pair.B).LoadAsync("eq/alarm-2", Ct);
return a is not null && b is not null
&& a.Comments.Count == 1 && b.Comments.Count == 1
&& a.Comments[0] == b.Comments[0];
},
"both nodes to agree on a single copy of the shared entry");
// Held, not merely reached: nothing late may double the trail on either node.
(await LocalDbPairHarness.HeldWithinAsync(
async () => await CommentCountAsync(pair.A, "eq/alarm-2") != 1
|| await CommentCountAsync(pair.B, "eq/alarm-2") != 1,
TimeSpan.FromSeconds(2)))
.ShouldBeFalse("the shared entry must stay one entry on both nodes, never two");
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs
+99 -10
View File
@@ -134,7 +134,7 @@ public sealed class LocalDbSetupTests : IDisposable
TestContext.Current.CancellationToken);
[Fact]
public void OnReady_WhenReplicationIsConfigured_RegistersExactlyTheFourReplicatedTables()
public void OnReady_WhenReplicationIsConfigured_RegistersExactlyTheFiveReplicatedTables()
{
// Both directions are load-bearing. "No fewer" catches a dropped RegisterReplicated call
// (that table then never replicates). "No more" catches an accidental registration —
@@ -142,11 +142,11 @@ public sealed class LocalDbSetupTests : IDisposable
var db = Boot(InitiatorReplication);
db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal)
.ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
.ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
}
[Fact]
public void OnReady_WhenOnlyTheSyncApiKeyIsSet_StillRegistersTheFourTables()
public void OnReady_WhenOnlyTheSyncApiKeyIsSet_StillRegistersTheFiveTables()
{
// The passive half of a pair carries the key and nothing else. If the predicate were
// "PeerAddress is set", this node would install no capture triggers and its OWN writes
@@ -155,7 +155,7 @@ public sealed class LocalDbSetupTests : IDisposable
var db = Boot(PassiveReplication);
db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal)
.ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
.ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
}
[Fact]
@@ -216,6 +216,56 @@ public sealed class LocalDbSetupTests : IDisposable
db.ReplicatedTables["alarm_condition_state"].PkColumns.ShouldBe(["scripted_alarm_id"]);
}
[Fact]
public void AlarmConditionComment_PkIsTheAlarmIdPlusTheTrailOrdinal()
{
// One row per trail entry. Keyed by ordinal (not content) so the pair's two copies of the
// same logical entry — each stamped by its own node's clock — collapse to one row under LWW.
var db = Boot(InitiatorReplication);
db.ReplicatedTables["alarm_condition_comment"].PkColumns.ShouldBe(["scripted_alarm_id", "seq"]);
}
[Fact]
public async Task OnReady_MovesAnOldShapeInlineCommentTrail_AfterRegistration_SoTheMoveReplicates()
{
// A file written by the pre-split store: the trail is inline on the condition row.
var first = Boot(InitiatorReplication);
await first.ExecuteAsync(
"""
INSERT INTO alarm_condition_state
(scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state,
comments_json, updated_at_utc)
VALUES ('eq/alarm-legacy', 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved',
'[{"timestampUtc":"2026-06-10T10:00:00Z","user":"jane","kind":"Acknowledge","text":"a"},{"timestampUtc":"2026-06-10T10:05:00Z","user":"bob","kind":"Confirm","text":"b"}]',
'2026-06-10T10:05:00.0000000Z')
""",
parameters: null,
TestContext.Current.CancellationToken);
// Positive control: before the restart the trail is inline and the comment table is empty.
(await ScalarAsync(first, "SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(0);
// The upgrade restart.
var db = Boot(InitiatorReplication);
(await ScalarAsync(db, "SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'eq/alarm-legacy'"))
.ShouldBe(2);
// The condition row itself is NOT rewritten (review m1): a migration UPDATE would re-stamp it
// with a fresh HLC and could beat a newer peer state under LWW. Only the seeding INSERT is in
// the oplog for it; the alarm's next save clears the inline column.
(await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_condition_state'"))
.ShouldBe(1);
// Captured: the migration ran after RegisterReplicated, so the moved entries reach the peer.
(await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'"))
.ShouldBe(2);
// Idempotent: a further boot writes nothing more.
var third = Boot(InitiatorReplication);
(await ScalarAsync(third, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'"))
.ShouldBe(2);
(await ScalarAsync(third, "SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(2);
}
[Fact]
public async Task AlarmSfEventRows_EnterTheOplog()
{
@@ -296,7 +346,7 @@ public sealed class LocalDbSetupTests : IDisposable
var replicated = Boot(InitiatorReplication);
await InsertPointerAsync(replicated, "SITE-A");
(await ScalarAsync(replicated, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'"))
.ShouldBe(12);
.ShouldBe(15);
var unreplicated = Boot();
@@ -350,6 +400,45 @@ public sealed class LocalDbSetupTests : IDisposable
sink.Events.Count(e => e.Level == LogEventLevel.Information).ShouldBe(1);
}
[Fact]
public async Task OnReady_WarnsOncePerAlarmWhoseInlineCommentTrailIsUnreadable()
{
// Review m2: an unmigratable history must be visible, not skipped silently on every boot.
var first = Boot();
await first.ExecuteAsync(
"""
INSERT INTO alarm_condition_state
(scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state,
comments_json, updated_at_utc)
VALUES (@Id, 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved', @Json, '2026-06-10T10:05:00.0000000Z')
""",
new { Id = "eq/alarm-bad", Json = "{not json" },
TestContext.Current.CancellationToken);
await first.ExecuteAsync(
"""
INSERT INTO alarm_condition_state
(scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state,
comments_json, updated_at_utc)
VALUES (@Id, 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved', @Json, '2026-06-10T10:05:00.0000000Z')
""",
new
{
Id = "eq/alarm-good",
Json = """[{"timestampUtc":"2026-06-10T10:00:00Z","user":"jane","kind":"AddComment","text":"a"}]""",
},
TestContext.Current.CancellationToken);
var sink = new CollectingSink();
Boot(replication: null, logger: BuildLogger(sink));
var warnings = sink.Events.Where(e => e.Level == LogEventLevel.Warning).ToList();
warnings.Count.ShouldBe(1);
warnings[0].RenderMessage().ShouldContain("eq/alarm-bad");
// Positive control: the readable row DID move, and said so.
sink.Events.Count(e => e.Level == LogEventLevel.Information && e.RenderMessage().Contains("copied"))
.ShouldBe(1);
}
[Fact]
public void DisablingReplication_SaysNothingWhenThereWasNothingToClean()
{
@@ -365,13 +454,13 @@ public sealed class LocalDbSetupTests : IDisposable
[Fact]
public void AFailureMidwayThroughRegistration_FailsBootLoudly_AndTheRetriedBootConverges()
{
// OnReady flips four tables in a loop with no try/catch, so a failure on the third leaves
// two registered and two not. That is the RIGHT behaviour and this pins both halves of why:
// OnReady flips five tables in a loop with no try/catch, so a failure on the third leaves
// two registered and three not. That is the RIGHT behaviour and this pins both halves of why:
// the throw propagates out of the AddZbLocalDb singleton factory and fails host start
// (a node half-capturing is a node whose replication is silently partial), and because both
// library APIs are idempotent the retried boot completes the flip rather than compounding it.
//
// Swallowing the throw would leave the node up, serving, and replicating two of four tables
// Swallowing the throw would leave the node up, serving, and replicating two of five tables
// with nothing in the log to say so.
var real = Boot(InitiatorReplication);
@@ -387,11 +476,11 @@ public sealed class LocalDbSetupTests : IDisposable
// pinning some unrelated throw, and the "partial flip" it claims to model never happened.
faulty.Calls.ShouldBe(3);
// The retry: same file, no wrapper. Idempotent re-registration converges on all four.
// The retry: same file, no wrapper. Idempotent re-registration converges on all five.
LocalDbSetup.OnReady(real, configuration);
real.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal)
.ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
.ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
}
/// <summary>
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs
+1 -1
View File
@@ -108,7 +108,7 @@ public sealed class LocalDbWiringTests : IDisposable
// Exact set, both directions — a dropped registration replicates nothing; an extra one costs
// three triggers and oplog volume on every write.
first.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal)
.ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
.ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
}
[Fact]
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs
+523
View File
@@ -0,0 +1,523 @@
using System.Collections.Immutable;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.ScriptedAlarms;
/// <summary>
/// The alarm comment audit trail lives in its own append-only table
/// (<see cref="AlarmConditionStateSchema.CommentsTable"/>) keyed by alarm id, so the condition
/// row stays a fixed size and a transition's write — and therefore its replication payload —
/// does not grow with the alarm's history.
/// </summary>
/// <remarks>
/// The payload is measured where it costs: the bytes the LocalDb capture triggers write into
/// <c>__localdb_oplog</c> for one save, which is exactly what ships to the pair peer. Both
/// tables are registered for capture here, as they are on a replicating node.
/// </remarks>
public sealed class LocalDbAlarmCommentAuditTests : IDisposable
{
private static readonly DateTime T0 = new(2026, 06, 10, 12, 00, 00, DateTimeKind.Utc);
private readonly string _dbPath =
Path.Combine(Path.GetTempPath(), $"otopcua-alarm-comments-{Guid.NewGuid():N}.db");
private readonly ServiceProvider _provider;
private readonly ILocalDb _db;
public LocalDbAlarmCommentAuditTests()
{
var configuration = new ConfigurationBuilder()
.AddInMemoryCollection(new Dictionary<string, string?> { ["LocalDb:Path"] = _dbPath })
.Build();
_provider = new ServiceCollection()
.AddZbLocalDb(configuration, db =>
{
using (var connection = db.CreateConnection())
{
AlarmConditionStateSchema.Apply(connection);
}
db.RegisterReplicated(AlarmConditionStateSchema.StateTable);
db.RegisterReplicated(AlarmConditionStateSchema.CommentsTable);
})
.BuildServiceProvider();
_db = _provider.GetRequiredService<ILocalDb>();
}
public void Dispose()
{
_provider.Dispose();
SqliteConnection.ClearAllPools();
foreach (var path in new[] { _dbPath, $"{_dbPath}-wal", $"{_dbPath}-shm" })
{
try { if (File.Exists(path)) File.Delete(path); } catch { /* best effort */ }
}
}
private LocalDbAlarmConditionStateStore NewStore()
=> new(_db, NullLogger<LocalDbAlarmConditionStateStore>.Instance);
private static CancellationToken Ct => CancellationToken.None;
private static ImmutableList<AlarmComment> Comments(int count, string prefix = "c")
{
var builder = ImmutableList.CreateBuilder<AlarmComment>();
for (var i = 0; i < count; i++)
builder.Add(new AlarmComment(T0.AddSeconds(i), "jane", "AddComment", $"{prefix}-{i:D6}"));
return builder.ToImmutable();
}
/// <summary>An unacknowledged state carrying <paramref name="comments"/>, ready to be acked.</summary>
private static AlarmConditionState Unacked(string alarmId, ImmutableList<AlarmComment> comments) =>
AlarmConditionState.Fresh(alarmId, T0) with
{
Active = AlarmActiveState.Active,
Acked = AlarmAckedState.Unacknowledged,
Comments = comments,
};
private async Task<long> OplogHighWaterAsync() =>
(await _db.QueryAsync("SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog", r => r.GetInt64(0), null, Ct))[0];
/// <summary>Bytes captured into the oplog above <paramref name="afterSeq"/> — the replication payload.</summary>
private async Task<long> OplogBytesAboveAsync(long afterSeq) =>
(await _db.QueryAsync(
"SELECT COALESCE(SUM(LENGTH(pk_json) + COALESCE(LENGTH(row_json), 0)), 0) FROM __localdb_oplog WHERE seq > @S",
r => r.GetInt64(0),
new { S = afterSeq },
Ct))[0];
private async Task<long> OplogEntriesAboveAsync(long afterSeq, string table) =>
(await _db.QueryAsync(
"SELECT COUNT(*) FROM __localdb_oplog WHERE seq > @S AND table_name = @T",
r => r.GetInt64(0),
new { S = afterSeq, T = table },
Ct))[0];
private async Task<long> ScalarAsync(string sql) =>
(await _db.QueryAsync(sql, r => r.GetInt64(0), null, Ct))[0];
/// <summary>
/// Seeds <paramref name="alarmId"/> with <paramref name="historyLength"/> comments, then
/// returns the oplog bytes one acknowledge transition (one new comment) costs.
/// </summary>
private async Task<long> AckTransitionBytesAsync(string alarmId, int historyLength, string ackComment)
{
var store = NewStore();
var seeded = Unacked(alarmId, Comments(historyLength));
await store.SaveAsync(seeded, Ct);
var before = await OplogHighWaterAsync();
var acked = Part9StateMachine.ApplyAcknowledge(seeded, "bob", ackComment, T0.AddDays(1)).State;
await store.SaveAsync(acked, Ct);
return await OplogBytesAboveAsync(before);
}
[Fact]
public async Task TransitionPayload_IsConstant_WhetherTheAlarmCarries10OrTenThousandComments()
{
var small = await AckTransitionBytesAsync("alarm-small", 10, "ack");
var large = await AckTransitionBytesAsync("alarm-large", 10_000, "ack");
// Constant: the only difference allowed is the width of the new comment's ordinal ("10" vs
// "10000", in pk_json and row_json) — a handful of bytes, not O(n). The alarm ids are the
// same length on purpose.
small.ShouldBeGreaterThan(0);
Math.Abs(large - small).ShouldBeLessThanOrEqualTo(16,
$"one ack transition cost {small} bytes at 10 comments but {large} at 10,000");
large.ShouldBeLessThan(2_048);
// Positive control on the instrument: the seeded history really is large and really was
// captured — otherwise a "constant" result could just mean capture is off.
(await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-large'"))
.ShouldBe(10_001);
var historyBytes = await ScalarAsync(
"SELECT SUM(LENGTH(row_json)) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'");
historyBytes.ShouldBeGreaterThan(1_000_000);
}
[Fact]
public async Task TransitionPayload_PositiveControl_TheMeasurementSeesTheNewCommentsBytes()
{
// Proves the measurement above is not vacuous: the oplog delta of a transition does carry
// the appended comment, so a 4 KB comment costs at least 4 KB.
var plain = await AckTransitionBytesAsync("alarm-plain", 10, "ack");
var heavy = await AckTransitionBytesAsync("alarm-heavy", 10, new string('x', 4_096));
(heavy - plain).ShouldBeGreaterThanOrEqualTo(4_096 * 2 - 16,
"the comment text rides both the comment row and last_ack_comment on the condition row");
}
[Fact]
public async Task Transition_WritesOneConditionRowAndOnlyTheAppendedCommentRow()
{
var store = NewStore();
var seeded = Unacked("alarm-append", Comments(50));
await store.SaveAsync(seeded, Ct);
var before = await OplogHighWaterAsync();
var acked = Part9StateMachine.ApplyAcknowledge(seeded, "bob", "ack", T0.AddDays(1)).State;
await store.SaveAsync(acked, Ct);
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(1);
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(1);
var loaded = await store.LoadAsync("alarm-append", Ct);
loaded.ShouldNotBeNull();
loaded.Comments.ShouldBe(acked.Comments);
}
[Fact]
public async Task ConditionRow_StaysFixedSize_InlineColumnIsAlwaysEmpty()
{
var store = NewStore();
await store.SaveAsync(Unacked("alarm-fixed", Comments(500)), Ct);
(await _db.QueryAsync(
"SELECT comments_json FROM alarm_condition_state WHERE scripted_alarm_id = 'alarm-fixed'",
r => r.GetString(0), null, Ct))[0]
.ShouldBe("[]");
(await store.LoadAsync("alarm-fixed", Ct))!.Comments.Count.ShouldBe(500);
}
[Fact]
public async Task UnchangedSave_WritesNoCommentRows()
{
var store = NewStore();
var state = Unacked("alarm-idle", Comments(20));
await store.SaveAsync(state, Ct);
var before = await OplogHighWaterAsync();
await store.SaveAsync(state with { LastTransitionUtc = T0.AddMinutes(1) }, Ct);
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(0);
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(1,
"positive control: the save itself was captured");
}
[Fact]
public async Task LoadAll_ReturnsEachAlarmsOwnTrailInOrder()
{
var store = NewStore();
var a = Unacked("alarm-a", Comments(3, "a"));
var b = Unacked("alarm-b", Comments(5, "b"));
await store.SaveAsync(a, Ct);
await store.SaveAsync(b, Ct);
var all = (await store.LoadAllAsync(Ct)).ToDictionary(s => s.AlarmId);
all["alarm-a"].Comments.ShouldBe(a.Comments);
all["alarm-b"].Comments.ShouldBe(b.Comments);
}
[Fact]
public async Task Remove_DeletesTheAlarmsTrail_AndAReCreatedAlarmStartsEmpty()
{
var store = NewStore();
await store.SaveAsync(Unacked("alarm-gone", Comments(4)), Ct);
await store.SaveAsync(Unacked("alarm-kept", Comments(2)), Ct);
await store.RemoveAsync("alarm-gone", Ct);
(await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-gone'"))
.ShouldBe(0);
(await store.LoadAsync("alarm-kept", Ct))!.Comments.Count.ShouldBe(2, "positive control");
var recreated = AlarmConditionState.Fresh("alarm-gone", T0)
with { Comments = ImmutableList.Create(new AlarmComment(T0, "amy", "AddComment", "new life")) };
await store.SaveAsync(recreated, Ct);
(await store.LoadAsync("alarm-gone", Ct))!.Comments.ShouldBe(recreated.Comments);
}
[Fact]
public async Task CommentsAlreadyHeldAtAnOrdinal_AreNeverRewritten()
{
// Append-only: a save whose list disagrees with what is stored at an existing ordinal (the
// peer's copy of the same logical entry already arrived) leaves the stored entry alone.
var store = NewStore();
var original = Unacked("alarm-ao", Comments(3));
await store.SaveAsync(original, Ct);
var rewritten = original with { Comments = original.Comments.SetItem(1, new AlarmComment(T0, "eve", "AddComment", "forged")) };
await store.SaveAsync(rewritten, Ct);
// The stored entry at ordinal 1 is untouched; the unrecognised local entry is kept by
// appending it (never dropped), not by overwriting history.
var loaded = (await store.LoadAsync("alarm-ao", Ct))!.Comments;
loaded.Take(3).ShouldBe(original.Comments);
loaded.Count.ShouldBe(4);
loaded[3].Text.ShouldBe("forged");
}
// ---- back-compat + migration -----------------------------------------------------------
/// <summary>
/// The exact shape the pre-split store wrote into <c>comments_json</c>
/// (<c>JsonSerializerDefaults.Web</c>: camelCase, Utc timestamps with a trailing <c>Z</c>).
/// </summary>
private const string OldShapeCommentsJson =
"""[{"timestampUtc":"2026-06-10T10:00:00Z","user":"jane","kind":"Acknowledge","text":"checking pump"},{"timestampUtc":"2026-06-10T11:30:00.1234567Z","user":"bob","kind":"Confirm","text":"all clear"},{"timestampUtc":"2026-06-10T11:30:00.1234567Z","user":"system","kind":"AutoUnshelve","text":""}]""";
private static readonly ImmutableList<AlarmComment> OldShapeComments = ImmutableList.Create(
new AlarmComment(new DateTime(2026, 06, 10, 10, 00, 00, DateTimeKind.Utc), "jane", "Acknowledge", "checking pump"),
new AlarmComment(new DateTime(2026, 06, 10, 11, 30, 00, DateTimeKind.Utc).AddTicks(1_234_567), "bob", "Confirm", "all clear"),
new AlarmComment(new DateTime(2026, 06, 10, 11, 30, 00, DateTimeKind.Utc).AddTicks(1_234_567), "system", "AutoUnshelve", ""));
private async Task InsertOldShapeRowAsync(string alarmId, string commentsJson = OldShapeCommentsJson) =>
await _db.ExecuteAsync(
"""
INSERT INTO alarm_condition_state
(scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state,
shelving_expires_utc, last_ack_user, last_ack_comment, last_ack_utc,
last_confirm_user, last_confirm_comment, last_confirm_utc, comments_json, updated_at_utc)
VALUES
(@Id, 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved',
NULL, 'jane', 'checking pump', '2026-06-10T10:00:00.0000000Z',
'bob', 'all clear', '2026-06-10T11:30:00.1234567Z', @Json, '2026-06-10T11:30:00.1234567Z')
""",
new { Id = alarmId, Json = commentsJson },
Ct);
private static void ShouldBeTheOldShapeState(AlarmConditionState? loaded, string alarmId)
{
loaded.ShouldNotBeNull();
loaded.AlarmId.ShouldBe(alarmId);
loaded.Acked.ShouldBe(AlarmAckedState.Acknowledged);
loaded.Confirmed.ShouldBe(AlarmConfirmedState.Confirmed);
loaded.LastAckUser.ShouldBe("jane");
loaded.LastConfirmComment.ShouldBe("all clear");
loaded.Comments.ShouldBe(OldShapeComments);
loaded.Comments.ShouldAllBe(c => c.TimestampUtc.Kind == DateTimeKind.Utc);
}
[Fact]
public async Task OldShapeRow_WithInlineComments_LoadsExactlyAsBefore()
{
await InsertOldShapeRowAsync("alarm-legacy");
ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-legacy", Ct), "alarm-legacy");
ShouldBeTheOldShapeState((await NewStore().LoadAllAsync(Ct)).Single(), "alarm-legacy");
}
[Fact]
public async Task Migration_CopiesInlineCommentsToTheAuditTable_AndNeverRewritesTheConditionRow()
{
await InsertOldShapeRowAsync("alarm-m1");
await InsertOldShapeRowAsync("alarm-m2");
await InsertOldShapeRowAsync("alarm-empty", "[]");
var before = await OplogHighWaterAsync();
AlarmConditionStateSchema.InlineCommentMigrationResult result;
using (var connection = _db.CreateConnection())
result = AlarmConditionStateSchema.MigrateInlineComments(connection);
result.RowsMoved.ShouldBe(2);
result.UnparseableAlarmIds.ShouldBeEmpty();
(await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(6);
// Review m1: the migration must not touch the condition row. A full-row UPDATE re-stamps it
// with a fresh HLC and can revert a newer peer state (e.g. an ack) under LWW. The next save
// (the engine's boot save) resets the column instead.
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(0);
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(6,
"positive control: the copied entries were captured");
ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-m1", Ct), "alarm-m1");
ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-m2", Ct), "alarm-m2");
(await NewStore().LoadAsync("alarm-empty", Ct))!.Comments.ShouldBeEmpty();
}
[Fact]
public async Task Migration_IsIdempotent_ASecondRunChangesNothing()
{
await InsertOldShapeRowAsync("alarm-idem");
using (var connection = _db.CreateConnection())
AlarmConditionStateSchema.MigrateInlineComments(connection).RowsMoved.ShouldBe(1, "positive control");
var before = await OplogHighWaterAsync();
using (var connection = _db.CreateConnection())
AlarmConditionStateSchema.MigrateInlineComments(connection).RowsMoved.ShouldBe(0);
(await OplogHighWaterAsync()).ShouldBe(before, "a re-run must write nothing");
(await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(3);
ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-idem", Ct), "alarm-idem");
}
[Fact]
public async Task Migration_OverAPartiallyMigratedRow_FillsOnlyTheMissingOrdinals()
{
// Only ordinal 0 made it across (e.g. a restored backup): the run fills 1 and 2, keeps 0,
// duplicates nothing.
await InsertOldShapeRowAsync("alarm-partial");
await _db.ExecuteAsync(
AlarmConditionStateSchema.InsertCommentSql,
new
{
Id = "alarm-partial", Seq = 0,
Ts = AlarmConditionStateSchema.FormatTimestamp(OldShapeComments[0].TimestampUtc),
User = OldShapeComments[0].User, Kind = OldShapeComments[0].Kind, Text = OldShapeComments[0].Text,
},
Ct);
using (var connection = _db.CreateConnection())
AlarmConditionStateSchema.MigrateInlineComments(connection).RowsMoved.ShouldBe(1);
(await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(3);
ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-partial", Ct), "alarm-partial");
}
[Fact]
public async Task Migration_ReportsAnUnparseableInlineTrail_AndLeavesItAlone()
{
// Review m2: an unreadable trail must surface (OnReady logs a Warning per id), not be
// skipped silently on every boot.
await InsertOldShapeRowAsync("alarm-bad", "{not json");
await InsertOldShapeRowAsync("alarm-good");
AlarmConditionStateSchema.InlineCommentMigrationResult result;
using (var connection = _db.CreateConnection())
result = AlarmConditionStateSchema.MigrateInlineComments(connection);
result.UnparseableAlarmIds.ShouldBe(["alarm-bad"]);
result.RowsMoved.ShouldBe(1, "positive control: the good row still moved");
(await _db.QueryAsync(
"SELECT comments_json FROM alarm_condition_state WHERE scripted_alarm_id = 'alarm-bad'",
r => r.GetString(0), null, Ct))[0]
.ShouldBe("{not json");
}
// ---- divergence / failover (review I-1) -------------------------------------------------
/// <summary>Captures log entries so a divergence Warning can be asserted, not inferred.</summary>
private sealed class CapturingLogger : Microsoft.Extensions.Logging.ILogger<LocalDbAlarmConditionStateStore>
{
public List<(Microsoft.Extensions.Logging.LogLevel Level, string Message)> Entries { get; } = [];
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true;
public void Log<TState>(
Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId,
TState state, Exception? exception, Func<TState, Exception?, string> formatter)
{
lock (Entries) Entries.Add((logLevel, formatter(state, exception)));
}
public int Warnings => Entries.Count(e => e.Level == Microsoft.Extensions.Logging.LogLevel.Warning);
}
[Fact]
public async Task Failover_TheSurvivorsNextEntry_IsPersisted_AfterItMissedACommandThePeerApplied()
{
// The I-1 sequence on node B's database:
// 1. a shared trail of 5 entries;
// 2. an AddComment reaches only node A (B's command delivery failed once) and A's row at
// ordinal 5 replicates into B's table — B's IN-MEMORY trail still has 5 entries;
// 3. A dies and the operator acks on B: B's in-memory Ack sits at index 5, colliding with
// A's entry at ordinal 5.
// The Ack must be persisted (and say so), not silently treated as "already stored".
var shared = Unacked("alarm-fo", Comments(5));
var a = NewStore();
await a.SaveAsync(shared, Ct);
var onlyOnA = Part9StateMachine.ApplyAddComment(shared, "jane", "seen on A only", T0.AddHours(1)).State;
await a.SaveAsync(onlyOnA, Ct);
var log = new CapturingLogger();
var b = new LocalDbAlarmConditionStateStore(_db, log);
var ackOnB = Part9StateMachine.ApplyAcknowledge(shared, "bob", "acked on B", T0.AddHours(2)).State;
await b.SaveAsync(ackOnB, Ct);
var loaded = (await NewStore().LoadAsync("alarm-fo", Ct))!;
loaded.Comments.Count.ShouldBe(7);
loaded.Comments[5].ShouldBe(onlyOnA.Comments[5]);
loaded.Comments[6].ShouldBe(ackOnB.Comments[5]);
log.Warnings.ShouldBe(1, "a divergence is never silent");
// Re-saving with no new entry must not append the Ack a second time (its own earlier
// divergence-append shifted it off its in-memory index).
var before = await OplogHighWaterAsync();
await b.SaveAsync(ackOnB with { LastTransitionUtc = T0.AddHours(3) }, Ct);
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(0);
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(1,
"positive control: the re-save itself was captured");
// And B's following entry (still one behind in memory) is persisted too, exactly once.
var confirmOnB = Part9StateMachine.ApplyAddComment(ackOnB, "bob", "follow-up on B", T0.AddHours(4)).State;
await b.SaveAsync(confirmOnB, Ct);
await b.SaveAsync(confirmOnB, Ct);
var after = (await NewStore().LoadAsync("alarm-fo", Ct))!;
after.Comments.Count.ShouldBe(8);
after.Comments[7].ShouldBe(confirmOnB.Comments[6]);
}
[Fact]
public async Task PeersCopyOfTheSameCommand_IsNotAppendedAgain_AndLogsNothing()
{
// The positive control for the divergence path: in normal operation both nodes apply every
// command, each with its own clock. When the peer's copy is already at the ordinal, the
// local copy is the same entry and must NOT be appended — otherwise the fix would double
// every entry on a healthy pair.
var shared = Unacked("alarm-dual", Comments(5));
await NewStore().SaveAsync(shared, Ct);
var ackOnA = Part9StateMachine.ApplyAcknowledge(shared, "bob", "on it", T0.AddHours(1)).State;
await NewStore().SaveAsync(ackOnA, Ct);
var log = new CapturingLogger();
var b = new LocalDbAlarmConditionStateStore(_db, log);
var ackOnB = Part9StateMachine.ApplyAcknowledge(shared, "bob", "on it", T0.AddHours(1).AddMilliseconds(40)).State;
var before = await OplogHighWaterAsync();
await b.SaveAsync(ackOnB, Ct);
(await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(0);
(await NewStore().LoadAsync("alarm-dual", Ct))!.Comments.Count.ShouldBe(6);
log.Warnings.ShouldBe(0);
}
[Fact]
public async Task ARepeatedIdenticalComment_IsKept_AsItsOwnEntry()
{
// Matching ignores the timestamp only within the clock-skew window; an operator repeating the
// same text later is a new entry, not a duplicate of the earlier one.
var store = NewStore();
var s = Unacked("alarm-repeat", Comments(2));
s = Part9StateMachine.ApplyAddComment(s, "amy", "still high", T0.AddHours(1)).State;
await store.SaveAsync(s, Ct);
s = Part9StateMachine.ApplyAddComment(s, "amy", "still high", T0.AddHours(2)).State;
await store.SaveAsync(s, Ct);
(await NewStore().LoadAsync("alarm-repeat", Ct))!.Comments.ShouldBe(s.Comments);
}
[Fact]
public async Task SavingALoadedOldShapeState_MovesItsCommentsAndAppendsTheNewOne()
{
// The lazy path: an old-shape row the boot migration never saw (it was written before the
// first save on this build) still converges on the first save.
await InsertOldShapeRowAsync("alarm-lazy");
var store = NewStore();
var loaded = await store.LoadAsync("alarm-lazy", Ct);
var next = Part9StateMachine.ApplyAddComment(loaded!, "amy", "follow-up", T0.AddDays(2)).State;
await store.SaveAsync(next, Ct);
(await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-lazy'"))
.ShouldBe(4);
(await _db.QueryAsync(
"SELECT comments_json FROM alarm_condition_state WHERE scripted_alarm_id = 'alarm-lazy'",
r => r.GetString(0), null, Ct))[0]
.ShouldBe("[]");
(await NewStore().LoadAsync("alarm-lazy", Ct))!.Comments.ShouldBe(next.Comments);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs
+1
View File
@@ -48,6 +48,7 @@ public sealed class LocalDbAlarmConditionStateStoreTests : IDisposable
}
db.RegisterReplicated(AlarmConditionStateSchema.StateTable);
db.RegisterReplicated(AlarmConditionStateSchema.CommentsTable);
})
.BuildServiceProvider();