From 271fcc4e15ecf520607d67fedfc3915479adca88 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 21 Jul 2026 04:39:32 -0400 Subject: [PATCH] test(localdb): alarm S&F convergence scenarios Two driver nodes replicating the alarm buffer over the real loopback h2c transport, through the real fail-closed interceptor. This is the test the phase exists to pass: schema, registration, sink rewire and drain gate can all be individually green while a pair still fails to share the undelivered alarm history that is the whole point of moving the queue. Rows are written through the production sink rather than hand-rolled INSERTs, so the id derivation, column set and delete-on-ack semantics under test are the ones production uses. The scenarios cover a buffered burst converging with the oplog draining to zero, delivered rows being removed from BOTH nodes as tombstones (the no-redeliver-after-failover property), writes during a transport outage surviving the rejoin, and the same event accepted on both nodes collapsing to one row. The positive control found a weak assertion. With RegisterReplicated("alarm_sf_events") commented out, three of the four went red immediately -- but the same-event-on-both-nodes scenario still PASSED, because with replication off each node trivially holds its own single copy and "one row on each" is satisfied by two databases that never spoke. It now also enqueues a distinct event on B and requires both nodes to hold two rows, which cannot be satisfied without convergence; it goes red under the control like the others. Control restored, all four green. Also updates the second exact-set replicated-tables pin, in LocalDbWiringTests. The plan predicted these pins would go red here, and that is them working: one of them is asserted through the full driver DI graph rather than the OnReady callback, so it catches a registration that exists in the callback but never reaches a running host. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- ...7-20-localdb-adoption-phase2.md.tasks.json | 5 +- .../LocalDb/AlarmSfConvergenceTests.cs | 197 ++++++++++++++++++ .../LocalDbWiringTests.cs | 4 +- 3 files changed, 202 insertions(+), 4 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfConvergenceTests.cs diff --git a/docs/plans/2026-07-20-localdb-adoption-phase2.md.tasks.json b/docs/plans/2026-07-20-localdb-adoption-phase2.md.tasks.json index 512b83cc..7e93a5bd 100644 --- a/docs/plans/2026-07-20-localdb-adoption-phase2.md.tasks.json +++ b/docs/plans/2026-07-20-localdb-adoption-phase2.md.tasks.json @@ -46,10 +46,11 @@ { "id": 5, "subject": "Task 5: Convergence + failover scenarios in the pair harness (+ positive control)", - "status": "pending", + "status": "completed", "blockedBy": [ 3 - ] + ], + "note": "4 scenarios green. POSITIVE CONTROL RUN (RegisterReplicated(alarm_sf_events) commented out): 3/4 went red immediately; the 4th (same-event-on-both-nodes) passed VACUOUSLY - with replication off each node trivially held its own single row. Strengthened by enqueuing a second distinct event on B and asserting BOTH nodes hold 2; it then went red under the control too. Control restored, all 4 green. Also fixed a SECOND exact-set pin the plan predicted: LocalDbWiringTests." }, { "id": 6, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfConvergenceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfConvergenceTests.cs new file mode 100644 index 00000000..2ebf485a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfConvergenceTests.cs @@ -0,0 +1,197 @@ +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb; + +/// +/// LocalDb Phase 2 — two driver nodes replicating the alarm store-and-forward buffer over a real +/// loopback h2c transport, through the real fail-closed interceptor. +/// +/// +/// +/// This is the test the phase exists to pass. Everything upstream — the schema, the +/// registration, the sink rewire, the drain gate — can be individually green while a +/// redundant pair still fails to share the undelivered alarm history that is the whole +/// point of moving the queue here. +/// +/// +/// Rows are written through the production rather +/// than by hand-rolled INSERTs. The id derivation, the column set and the delete-on-ack +/// semantics under test are then the ones production actually uses. +/// +/// +[Collection("LocalDbPairConvergence")] +public sealed class AlarmSfConvergenceTests +{ + private static readonly Serilog.ILogger Log = new LoggerConfiguration().CreateLogger(); + + /// A writer that acknowledges everything — the drain's happy path. + private sealed class AckWriter : IAlarmHistorianWriter + { + public Task> WriteBatchAsync( + IReadOnlyList batch, CancellationToken cancellationToken) => + Task.FromResult>( + batch.Select(_ => HistorianWriteOutcome.Ack).ToList()); + } + + /// A writer that never acknowledges — a historian outage, which is when the buffer matters. + private sealed class OutageWriter : IAlarmHistorianWriter + { + public Task> WriteBatchAsync( + IReadOnlyList batch, CancellationToken cancellationToken) => + Task.FromResult>( + batch.Select(_ => HistorianWriteOutcome.RetryPlease).ToList()); + } + + private static AlarmHistorianEvent Event(string alarmId, int second) => new( + AlarmId: alarmId, + EquipmentPath: "line-1/eq-1", + AlarmName: "temp-high", + AlarmTypeName: "LimitAlarm", + Severity: AlarmSeverity.High, + EventKind: "Activated", + Message: "temperature high", + User: "system", + Comment: null, + TimestampUtc: new DateTime(2026, 7, 21, 0, 0, second, DateTimeKind.Utc)); + + /// The set of row ids on a node, ordered — equal sets on both nodes means converged. + private static async Task IdsAsync(ILocalDb db) + { + var rows = await db.QueryAsync( + "SELECT id FROM alarm_sf_events ORDER BY id", + static r => r.GetString(0)); + return string.Join(",", rows); + } + + private static async Task CountAsync(ILocalDb db) + { + var rows = await db.QueryAsync( + "SELECT COUNT(*) FROM alarm_sf_events", static r => r.GetInt64(0)); + return rows[0]; + } + + [Fact] + public async Task AlarmBurstOnA_ConvergesToB_AndOplogDrains() + { + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + // The Primary's sink. Its drain is left unstarted so the burst stays queued and observable; + // an outage is exactly the state in which this buffer has to survive a node loss. + using var sink = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log); + for (var i = 0; i < 25; i++) + await sink.EnqueueAsync(Event($"eq/alarm-{i}", i), TestContext.Current.CancellationToken); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CountAsync(pair.B) == 25, + "node B to hold all 25 buffered alarm events"); + + (await IdsAsync(pair.B)).ShouldBe(await IdsAsync(pair.A)); + + // Oplog drains to empty once B acks: the steady state of a converged pair, and the proof + // the rows were genuinely acknowledged rather than merely sent. + await LocalDbPairHarness.WaitUntilAsync( + async () => await LocalDbPairHarness.OplogDepthAsync(pair.A) == 0, + "node A's oplog to drain after node B acknowledges the burst"); + } + + [Fact] + public async Task DeliveredRowsAreRemovedFromBothNodes() + { + // The no-redeliver-after-failover property, asserted at the data layer. An acknowledged row + // is DELETEd, and the delete has to reach the peer as a tombstone — otherwise a promoted + // standby would find the Primary's already-delivered events still queued and send them again. + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + using var buffering = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log); + for (var i = 0; i < 5; i++) + await buffering.EnqueueAsync(Event($"eq/alarm-{i}", i), TestContext.Current.CancellationToken); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CountAsync(pair.B) == 5, + "node B to receive the buffered events"); + + // The historian comes back and the Primary drains. + using var draining = new LocalDbStoreAndForwardSink(pair.A, new AckWriter(), Log); + await draining.DrainOnceAsync(TestContext.Current.CancellationToken); + (await CountAsync(pair.A)).ShouldBe(0); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CountAsync(pair.B) == 0, + "node B to drop its copies once node A delivered them"); + + (await LocalDbPairHarness.TombstoneCountAsync(pair.B, "alarm_sf_events")) + .ShouldBe(5, "the deletes must arrive as tombstones, not vanish silently"); + } + + [Fact] + public async Task WritesWhileTransportDown_SurviveRejoin() + { + // The realistic outage shape: alarms keep arriving on the Primary while the pair link is + // down. Nothing may be lost when it comes back. + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + using var sink = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log); + await sink.EnqueueAsync(Event("eq/before", 1), TestContext.Current.CancellationToken); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CountAsync(pair.B) == 1, + "the pre-outage event to reach node B"); + + await pair.StopPassiveAsync(); + + for (var i = 0; i < 10; i++) + await sink.EnqueueAsync(Event($"eq/during-{i}", 10 + i), TestContext.Current.CancellationToken); + + await pair.RestartPairAsync(); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CountAsync(pair.B) == 11, + "every event buffered during the outage to reach node B after the rejoin"); + + (await IdsAsync(pair.B)).ShouldBe(await IdsAsync(pair.A)); + } + + [Fact] + public async Task TheSameEventOnBothNodes_ConvergesToOneRow() + { + // Both adapters default-write while their redundancy role is unknown, so in the boot window + // the pair genuinely double-accepts the same fanned transition. Payload-derived ids are what + // turn that into one row instead of two; random GUIDs would leave the duplicate in the + // buffer for the eventual Primary to deliver twice. + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + var shared = Event("eq/shared", 1); + using var sinkA = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log); + using var sinkB = new LocalDbStoreAndForwardSink(pair.B, new OutageWriter(), Log); + + await sinkA.EnqueueAsync(shared, TestContext.Current.CancellationToken); + await sinkB.EnqueueAsync(shared, TestContext.Current.CancellationToken); + + // A second, DISTINCT event on B. Without it the scenario would pass with replication + // switched off entirely — each node would hold its own single copy and "count == 1 on + // both" would be satisfied by two unconverged databases. + await sinkB.EnqueueAsync(Event("eq/only-on-b", 2), TestContext.Current.CancellationToken); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CountAsync(pair.A) == 2 && await CountAsync(pair.B) == 2, + "both nodes to hold two rows: the doubly-accepted event collapsed, plus B's own"); + + (await IdsAsync(pair.A)).ShouldBe(await IdsAsync(pair.B)); + + // Held, not merely reached: a late-arriving replica of the peer's copy of the shared event + // must not add a third row a moment later. + (await LocalDbPairHarness.HeldWithinAsync( + async () => await CountAsync(pair.A) > 2 || await CountAsync(pair.B) > 2, + TimeSpan.FromSeconds(3))) + .ShouldBeFalse("the duplicate must stay collapsed"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs index 77060548..82f6c36e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs @@ -84,7 +84,7 @@ public sealed class LocalDbWiringTests : IDisposable } [Fact] - public void DriverGraph_LocalDbResolvesAsASingletonWithBothCacheTablesRegistered() + public void DriverGraph_LocalDbResolvesAsASingletonWithEveryReplicatedTableRegistered() { var sp = BuildDriverGraph(); @@ -95,7 +95,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(["deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } [Fact]