fix(historian): preserve stable alarm delivery identities
v2-ci / build (pull_request) Successful in 5m52s
v2-ci / unit-tests (pull_request) Failing after 1h23m55s

This commit is contained in:
Joseph Doherty
2026-10-13 07:10:58 -04:00
parent 0ccfbbee5c
commit 109163621c
13 changed files with 341 additions and 27 deletions
@@ -0,0 +1,78 @@
using System.Text.Json;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests;
public sealed class AlarmHistorianEventIdentityTests
{
internal const string Payload = """{"AlarmId":"A","EquipmentPath":"S","AlarmName":"N","AlarmTypeName":"LimitAlarm","Severity":2,"EventKind":"Activated","Message":"m","User":"system","Comment":null,"TimestampUtc":"2026-01-01T00:00:00Z"}""";
internal const string QueueId = "52C8292271C629CCC2BFE4EC8C16A5E70CA7456C7D88BC282ABCEDF636F05901";
private static AlarmHistorianEvent Event() => new("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
"Activated", "m", "system", null, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
[Theory]
[InlineData(QueueId, "48bca092-b99d-8ea4-a2de-8b30d80a481a")]
[InlineData("persisted-row-v1", "39473040-002b-876c-a1e6-9a9ad4a60ded")]
public void Persisted_keys_have_frozen_versioned_delivery_GUID_vectors(string key, string expected)
{
Assert.Equal(Guid.Parse(expected), AlarmDeliveryIds.FromQueueId(key));
}
[Fact]
public void Delivery_identity_is_nonzero_and_a_copy_preserves_it()
{
var original = Event();
Assert.NotEqual(Guid.Empty, original.DeliveryId);
Assert.Equal(original.DeliveryId, (original with { }).DeliveryId);
Assert.Equal(original.DeliveryId, (original with { Comment = "copy" }).DeliveryId);
}
[Fact]
public void Business_equality_hash_and_historical_JSON_ignore_delivery_identity()
{
var first = Event();
var second = Event();
Assert.NotEqual(first.DeliveryId, second.DeliveryId);
Assert.Equal(first, second);
Assert.True(first == second);
Assert.Equal(first.GetHashCode(), second.GetHashCode());
Assert.Single(new HashSet<AlarmHistorianEvent> { first, second });
Assert.Equal(Payload, JsonSerializer.Serialize(first));
Assert.Equal(Payload, JsonSerializer.Serialize(second));
Assert.Equal(QueueId, AlarmSfSchema.DeriveId(Payload));
Assert.Equal(first, JsonSerializer.Deserialize<AlarmHistorianEvent>(Payload));
}
[Theory]
[InlineData("AlarmId")]
[InlineData("EquipmentPath")]
[InlineData("AlarmName")]
[InlineData("AlarmTypeName")]
[InlineData("Severity")]
[InlineData("EventKind")]
[InlineData("Message")]
[InlineData("User")]
[InlineData("Comment")]
[InlineData("TimestampUtc")]
public void Each_business_field_still_participates_in_equality(string field)
{
var first = Event();
var changed = field switch
{
"AlarmId" => first with { AlarmId = "B" },
"EquipmentPath" => first with { EquipmentPath = "T" },
"AlarmName" => first with { AlarmName = "Other" },
"AlarmTypeName" => first with { AlarmTypeName = "DiscreteAlarm" },
"Severity" => first with { Severity = AlarmSeverity.Low },
"EventKind" => first with { EventKind = "Cleared" },
"Message" => first with { Message = "other" },
"User" => first with { User = "operator" },
"Comment" => first with { Comment = "note" },
"TimestampUtc" => first with { TimestampUtc = first.TimestampUtc.AddTicks(1) },
_ => throw new ArgumentOutOfRangeException(nameof(field)),
};
Assert.NotEqual(first, changed);
Assert.False(first == changed);
}
}
@@ -123,6 +123,69 @@ public sealed class LocalDbStoreAndForwardSinkTests : IDisposable
Comment: null,
TimestampUtc: ts ?? DateTime.UtcNow);
[Fact]
public async Task Queued_delivery_identity_survives_retry_reopen_and_payload_changes()
{
// The persisted row key is authoritative: reserializing today's payload must never change a retry's identity.
var expected = Guid.Parse("48bca092-b99d-8ea4-a2de-8b30d80a481a");
var evt = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
var writer = new FakeWriter { DefaultOutcome = HistorianWriteOutcome.RetryPlease };
using (var sink = new LocalDbStoreAndForwardSink(Db, writer, _log))
{
await sink.EnqueueAsync(evt, TestContext.Current.CancellationToken);
var row = (await Db.QueryAsync("SELECT id, payload_json FROM alarm_sf_events",
r => (Id: r.GetString(0), Payload: r.GetString(1)), null, TestContext.Current.CancellationToken)).Single();
row.Id.ShouldBe(AlarmHistorianEventIdentityTests.QueueId);
row.Payload.ShouldBe(AlarmHistorianEventIdentityTests.Payload);
await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
writer.Batches.ShouldAllBe(batch => batch.Single().DeliveryId == expected);
await Db.ExecuteAsync("UPDATE alarm_sf_events SET payload_json = @Payload",
new { Payload = System.Text.Json.JsonSerializer.Serialize(evt with { Comment = "edited" }) },
TestContext.Current.CancellationToken);
}
_provider!.Dispose();
_provider = null;
_db = null;
var reopenedWriter = new FakeWriter();
using var reopened = new LocalDbStoreAndForwardSink(Db, reopenedWriter, _log);
await reopened.DrainOnceAsync(TestContext.Current.CancellationToken);
reopenedWriter.Batches.Single().Single().DeliveryId.ShouldBe(expected);
reopenedWriter.Batches.Single().Single().Comment.ShouldBe("edited");
reopened.GetStatus().QueueDepth.ShouldBe(0);
}
[Fact]
public async Task Independently_created_identical_events_keep_the_historical_queue_key()
{
var first = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
var second = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
first.DeliveryId.ShouldNotBe(second.DeliveryId);
using var sink = new LocalDbStoreAndForwardSink(Db, new FakeWriter(), _log);
await sink.EnqueueAsync(first, TestContext.Current.CancellationToken);
await sink.EnqueueAsync(second, TestContext.Current.CancellationToken);
sink.GetStatus().QueueDepth.ShouldBe(1);
(await Db.QueryAsync("SELECT id FROM alarm_sf_events", r => r.GetString(0), null, TestContext.Current.CancellationToken))
.ShouldBe([AlarmHistorianEventIdentityTests.QueueId]);
}
[Fact]
public async Task Distinct_business_events_at_the_same_time_have_distinct_queued_delivery_ids()
{
var first = Event("A", new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
var second = first with { EventKind = "Cleared", Comment = "different transition" };
var writer = new FakeWriter();
using var sink = new LocalDbStoreAndForwardSink(Db, writer, _log);
await sink.EnqueueAsync(first, TestContext.Current.CancellationToken);
await sink.EnqueueAsync(second, TestContext.Current.CancellationToken);
sink.GetStatus().QueueDepth.ShouldBe(2);
await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
var delivered = writer.Batches.Single();
delivered.Count.ShouldBe(2);
delivered.ShouldAllBe(evt => evt.DeliveryId != Guid.Empty);
delivered.Select(evt => evt.DeliveryId).Distinct().Count().ShouldBe(2);
}
/// <summary>Verifies that acknowledged events are removed from the queue.</summary>
[Fact]
public async Task EnqueueThenDrain_Ack_removes_row()
@@ -297,6 +360,11 @@ public sealed class LocalDbStoreAndForwardSinkTests : IDisposable
var status = sink.GetStatus();
status.QueueDepth.ShouldBe(1);
status.DeadLetterDepth.ShouldBe(0);
var originalDeliveryId = writer.Batches.Single().Single().DeliveryId;
await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
writer.Batches.Count.ShouldBe(2);
writer.Batches.Last().Single().DeliveryId.ShouldBe(originalDeliveryId);
sink.GetStatus().QueueDepth.ShouldBe(0);
}
/// <summary>Verifies that the exponential backoff ladder caps at 60 seconds.</summary>
@@ -18,6 +18,33 @@ public sealed class GatewayAlarmHistorianWriterTests
private static GatewayAlarmHistorianWriter Writer(FakeHistorianGatewayClient fake) =>
new(fake, NullLogger<GatewayAlarmHistorianWriter>.Instance);
[Fact]
public async Task A_direct_retry_preserves_the_original_delivery_id()
{
var evt = Evt("A");
var fake = new FakeHistorianGatewayClient { SendEventThrows = new RpcException(new Status(StatusCode.Unavailable, "down")) };
using var writer = Writer(fake);
Assert.Equal(HistorianWriteOutcome.RetryPlease,
(await writer.WriteBatchAsync([evt], TestContext.Current.CancellationToken)).Single());
var first = fake.LastSendEvent!.Id;
fake.SendEventThrows = null;
fake.SendEventResult = new WriteAck { Success = true };
Assert.Equal(HistorianWriteOutcome.Ack,
(await writer.WriteBatchAsync([evt with { }], TestContext.Current.CancellationToken)).Single());
Assert.Equal(evt.DeliveryId.ToString("D"), first);
Assert.Equal(first, fake.LastSendEvent!.Id);
}
[Fact]
public async Task An_empty_delivery_id_is_permanent_failure_without_an_RPC()
{
var fake = new FakeHistorianGatewayClient();
using var writer = Writer(fake);
Assert.Equal(HistorianWriteOutcome.PermanentFail,
(await writer.WriteBatchAsync([Evt("A") with { DeliveryId = Guid.Empty }], TestContext.Current.CancellationToken)).Single());
Assert.Equal(0, fake.SendEventCallCount);
}
[Fact]
public async Task All_acked_when_SendEvent_succeeds()
{
@@ -149,9 +149,8 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
/// <see cref="GatewayAlarmHistorianWriter"/>, then <c>ReadEvents</c> the recent window for that
/// source through the real <see cref="GatewayHistorianDataSource"/> and assert the event is
/// present. Requires the gateway running <c>RuntimeDb:EventReadsEnabled=true</c> (the SQL
/// alarm-history read path). Presence is asserted as "at least one event for the source surfaced
/// in the post-send window" (the data source filters by source); the exact AlarmId / message
/// match is surfaced as a diagnostic, since the SQL event store may re-key the row.
/// alarm-history read path). The exact delivery GUID must return; an unrelated event for the
/// same source cannot satisfy the assertion.
/// </summary>
[Fact]
[Trait("Category", "LiveIntegration")]
@@ -196,7 +195,7 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
var read = await dataSource.ReadEventsAsync(
source, eventUtc - TimeSpan.FromMinutes(5), DateTime.UtcNow + TimeSpan.FromMinutes(1), maxEvents: 0, ct);
events = read.Events;
if (events.Count > 0) break;
if (events.Any(e => Guid.TryParse(e.EventId, out var id) && id == alarm.DeliveryId)) break;
await Task.Delay(TimeSpan.FromSeconds(2), ct);
}
while (DateTime.UtcNow < deadline);
@@ -211,18 +210,18 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
$"a source-filtered ReadEvents must return the just-sent alarm for source '{source}': the gateway now " +
"populates Source_Object from the event SourceName (C4 fix), so OtOpcUa's own alarm sends round-trip.");
var exactMatch = events.Any(e => string.Equals(e.EventId, alarmId, StringComparison.Ordinal));
var exactMatch = events.Any(e => Guid.TryParse(e.EventId, out var id) && id == alarm.DeliveryId);
exactMatch.ShouldBeTrue("the just-sent event must retain its nonzero delivery GUID through persistence and readback.");
TestContext.Current.SendDiagnosticMessage(
$"alarm round-trip: SendEvent source='{source}' id={alarmId} → ReadEvents returned {events.Count} event(s); exact-id match={exactMatch}.");
$"alarm round-trip: SendEvent source='{source}' delivery-id={alarm.DeliveryId:D} → exact ID returned among {events.Count} event(s).");
}
/// <summary>
/// Alarm send contract — the OtOpcUa responsibility in isolation: an <see cref="AlarmHistorianEvent"/>
/// maps to a wire event the gateway accepts and <c>SendEvent</c> returns <see cref="HistorianWriteOutcome.Ack"/>.
/// This is the half that must always hold (the readback half is historian-gated; see
/// <see cref="Alarm_SendEvent_then_ReadEvents"/>). Regression guard for the event-id mapping —
/// a client-supplied wire <c>Id</c> makes the gateway's SendEvent handler throw, so the mapper
/// must leave it unset.
/// <see cref="Alarm_SendEvent_then_ReadEvents"/>). The mapper sends the stable nonzero delivery GUID;
/// the earlier assumption that the gateway rejected every client-supplied ID was incorrect (#705).
/// </summary>
[Fact]
[Trait("Category", "LiveIntegration")]
@@ -251,7 +250,7 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
outcomes.ShouldHaveSingleItem().ShouldBe(
HistorianWriteOutcome.Ack,
"the alarm SendEvent must be acked (the AlarmEventMapper must NOT set the wire event Id — the gateway rejects a client-supplied id).");
"the alarm SendEvent carrying its nonzero delivery GUID must be acked.");
}
/// <summary>
@@ -7,6 +7,17 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.Mapping;
public sealed class AlarmEventMapperTests
{
[Fact]
public void A_direct_event_has_a_nonzero_wire_id_that_survives_repeated_mapping()
{
var alarm = new AlarmHistorianEvent("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
"Activated", "m", "system", null, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
var first = AlarmEventMapper.ToHistorianEvent(alarm).Id;
Assert.True(Guid.TryParse(first, out var id), $"Expected a delivery GUID; received '{first}'.");
Assert.NotEqual(Guid.Empty, id);
Assert.Equal(first, AlarmEventMapper.ToHistorianEvent(alarm).Id);
}
[Fact]
public void Maps_source_time_type_and_rich_properties()
{
@@ -17,9 +28,7 @@ public sealed class AlarmEventMapperTests
Assert.Equal("Area/Line/Pump1", e.SourceName);
Assert.Equal("LimitAlarm", e.Type);
Assert.Equal(new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc), e.EventTime.ToDateTime());
// The wire Id is deliberately left unset — the gateway's SendEvent rejects a client-supplied
// event id. The alarm's own id is carried as a property instead.
Assert.Equal(string.Empty, e.Id);
Assert.Equal(a.DeliveryId.ToString("D"), e.Id);
Assert.Equal("A1", e.Properties["AlarmId"]);
Assert.Equal("HiHi", e.Properties["AlarmName"]);
Assert.Equal("Activated", e.Properties["EventKind"]);
@@ -67,14 +76,21 @@ public sealed class AlarmEventMapperTests
}
[Fact]
public void Wire_id_is_never_set_and_blank_alarm_id_adds_no_property()
public void A_blank_condition_id_still_has_delivery_identity_and_adds_no_condition_property()
{
// A blank AlarmId must NOT fall back to a generated wire Id (the gateway rejects any id) and
// must not add an empty AlarmId property.
// Condition identity and event delivery identity serve different purposes.
var a = new AlarmHistorianEvent("", "S", "N", "DiscreteAlarm", AlarmSeverity.Low, "Cleared", "m", "system", null,
new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
var e = AlarmEventMapper.ToHistorianEvent(a);
Assert.Equal(string.Empty, e.Id);
Assert.Equal(a.DeliveryId.ToString("D"), e.Id);
Assert.False(e.Properties.ContainsKey("AlarmId"));
}
[Fact]
public void An_explicitly_empty_delivery_id_is_rejected_instead_of_reminted()
{
var alarm = new AlarmHistorianEvent("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
"Activated", "m", "system", null, DateTime.UtcNow) { DeliveryId = Guid.Empty };
Assert.Throws<ArgumentException>(() => AlarmEventMapper.ToHistorianEvent(alarm));
}
}
@@ -41,10 +41,14 @@ public sealed class AlarmSfConvergenceTests
/// <summary>A writer that never acknowledges — a historian outage, which is when the buffer matters.</summary>
private sealed class OutageWriter : IAlarmHistorianWriter
{
public List<AlarmHistorianEvent> Events { get; } = [];
public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken) =>
Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(
IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
{
Events.AddRange(batch);
return Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(
batch.Select(_ => HistorianWriteOutcome.RetryPlease).ToList());
}
}
private static AlarmHistorianEvent Event(string alarmId, int second) => new(
@@ -170,11 +174,15 @@ public sealed class AlarmSfConvergenceTests
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);
var sharedOnB = Event("eq/shared", 1);
shared.DeliveryId.ShouldNotBe(sharedOnB.DeliveryId);
var writerA = new OutageWriter();
var writerB = new OutageWriter();
using var sinkA = new LocalDbStoreAndForwardSink(pair.A, writerA, Log);
using var sinkB = new LocalDbStoreAndForwardSink(pair.B, writerB, Log);
await sinkA.EnqueueAsync(shared, TestContext.Current.CancellationToken);
await sinkB.EnqueueAsync(shared, TestContext.Current.CancellationToken);
await sinkB.EnqueueAsync(sharedOnB, 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
@@ -193,5 +201,13 @@ public sealed class AlarmSfConvergenceTests
async () => await CountAsync(pair.A) > 2 || await CountAsync(pair.B) > 2,
TimeSpan.FromSeconds(3)))
.ShouldBeFalse("the duplicate must stay collapsed");
// Either node can retry after promotion, with the same GUID despite independent input identities.
await sinkA.DrainOnceAsync(TestContext.Current.CancellationToken);
await sinkB.DrainOnceAsync(TestContext.Current.CancellationToken);
var deliveryA = writerA.Events.Single(e => e.AlarmId == "eq/shared").DeliveryId;
var deliveryB = writerB.Events.Single(e => e.AlarmId == "eq/shared").DeliveryId;
deliveryA.ShouldNotBe(Guid.Empty);
deliveryB.ShouldBe(deliveryA);
}
}
@@ -36,6 +36,38 @@ public sealed class AlarmSfLegacyMigratorTests : IDisposable
private ServiceProvider? _provider;
private sealed class CapturingRetryWriter : IAlarmHistorianWriter
{
public List<AlarmHistorianEvent> Events { get; } = [];
public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
{
Events.AddRange(batch);
return Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(batch.Select(_ => HistorianWriteOutcome.RetryPlease).ToArray());
}
}
[Fact]
public async Task A_legacy_payload_keeps_its_key_and_stable_delivery_identity_across_drains()
{
var payload = Payload("eq/a");
SeedLegacy((1, "eq/a", payload, 0, dead: 0));
var db = BuildDb();
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
var row = (await db.QueryAsync("SELECT id, payload_json FROM alarm_sf_events",
r => (Id: r.GetString(0), Payload: r.GetString(1)), null, TestContext.Current.CancellationToken)).Single();
row.Id.ShouldBe("B1D80B7207C3F3B6CB31105DBF1A6E0D6FE300DA1B3F877AB746B36535310A60");
row.Payload.ShouldBe(payload);
var writer = new CapturingRetryWriter();
using var log = new Serilog.LoggerConfiguration().CreateLogger();
using (var sink = new LocalDbStoreAndForwardSink(db, writer, log))
await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
using (var sink = new LocalDbStoreAndForwardSink(db, writer, log))
await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
writer.Events.Count.ShouldBe(2);
writer.Events.ShouldAllBe(evt => evt.DeliveryId == Guid.Parse("55372883-eaec-8dd7-a1a4-6f98f416a272"));
}
// ---- the cases ---------------------------------------------------------------------------
[Fact]