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

feat(mqtt): Sparkplug primary-host STATE publishing, Primary-gated (#511)

Browse Source 
`ActAsPrimaryHost` was accepted by config and by the AdminUI form and did nothing but
log a warning. It now delivers the whole three-part contract or none of it: a retained
`ONLINE` on `spBv1.0/STATE/{hostId}`, a matching retained `OFFLINE` armed as the MQTT
Last Will, and an explicit `OFFLINE` on a clean stand-down.

**The claim is a dedicated connection, not a message.** A Will can only be registered in
the CONNECT packet — it cannot be added to a live session, changed, or withdrawn. So a
claim that is acquired and released at runtime cannot ride the long-lived ingest session.
`SparkplugPrimaryHost` opens its own connection (client id `{clientId}-spbhost`, forced
clean-session) when the node gains Primary and closes it when the node loses Primary; the
connection's lifetime *is* the claim, so every leg follows from its lifecycle. Ingest is
untouched by a role change — no dropped subscriptions, no lost birth cache, no data gap at
the Secondary->Primary transition.

**Primary-gated** (owner decision, recorded on the issue). Both pair nodes subscribe
independently, so an ungated claim means both claim the same host id — and the sharp
failure is not the two fighting over the retained STATE, it is that the SECONDARY's death
fires its Will and publishes OFFLINE while the Primary is healthy, stopping every edge node
in the group. Registering the Will only on whichever node is Primary at connect time fails
the other way: a node that later gains Primary would publish ONLINE with no Will behind it.

New seam: `IDriverPrimaryRoleSource` (`Core.Abstractions`) carrying the *data-plane* verdict
(`PrimaryGatePolicy.ShouldServiceAsPrimary`), not the alarm-drain verdict — an unknown role
with a driver peer present must deny. Registered unconditionally by `AddOtOpcUaRuntime`,
deliberately NOT alongside `IRedundancyRoleView`, which `AddAlarmHistorian` registers only
when that section is enabled: piggybacking would leave a pair with no alarm historian
falling back to "claim unconditionally". Polled on a 1s timer rather than evented — a
process singleton raising events at per-instance drivers outlives every driver it holds.

Accepted trade: a brief edge-visible OFFLINE window during failover. That is a genuine
state transition, not a glitch to hide.

Live-gated 5/5 against the Mosquitto fixture, including the Will leg — `AbandonForTests`
drops the socket with no DISCONNECT, because a clean disconnect makes the broker discard
the Will by protocol rule, so a test driven through `StopAsync` would observe the
stand-down OFFLINE and pass against an implementation that never armed a Will. Falsified:
removing the Will turns that leg and only that leg red.

Also corrects two stale claims found en route: the AdminUI form still said MQTT tags are
read-only (#508 shipped), and the driver doc still said nothing consumes `IRediscoverable`.

Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
This commit is contained in:
Joseph Doherty
2026-07-31 07:17:17 -04:00
parent 2a7d11674f
commit 7f2108dcf2
21 changed files with 1881 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttPrimaryHostLiveTests.cs
+379
View File
@@ -0,0 +1,379 @@
using System.Buffers;
using System.Collections.Concurrent;
using MQTTnet;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;
/// <summary>
/// Live gate for the Sparkplug primary-host claim (Gitea #511), against the Mosquitto fixture.
/// </summary>
/// <remarks>
/// <para>
/// The issue's acceptance criteria, leg by leg: the broker holds a retained <c>ONLINE</c> while
/// the claim runs; an abrupt death leaves <c>OFFLINE</c> (the Will fires); a clean shutdown
/// leaves <c>OFFLINE</c>; and the pair-node question is answered rather than left to chance —
/// a node that is not Primary claims nothing, and a role change moves the claim.
/// </para>
/// <para>
/// <b>Each test uses its own host id.</b> A Sparkplug host id is a broker-wide singleton by
/// definition, and the retained STATE outlives the test that wrote it — two tests sharing an id
/// would see each other's leftovers, and would do so intermittently depending on ordering.
/// </para>
/// <para>
/// <b>Retained messages are cleaned up.</b> A retained <c>ONLINE</c> for a host that no longer
/// exists is exactly the lie this feature prevents, so leaving one on a shared fixture would be
/// poor manners at best and a confusing false positive for the next run at worst.
/// </para>
/// </remarks>
[Trait("Category", "Integration")]
[Collection("MqttFixture")]
public sealed class MqttPrimaryHostLiveTests(MqttFixture fixture)
{
private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(15);
private void SkipIfUnconfigured() => Assert.SkipWhen(fixture.SkipReason is not null, fixture.SkipReason ?? "");
[Fact]
public async Task A_primary_publishes_a_retained_ONLINE_that_a_late_subscriber_still_sees()
{
// Retention is the point, not an optimisation: an edge node that connects AFTER the host came
// up must still learn the host is alive. A non-retained ONLINE reaches only whoever happened to
// be subscribed at that instant, and every later joiner sees an absent host forever.
SkipIfUnconfigured();
var hostId = NewHostId();
var ct = TestContext.Current.CancellationToken;
await using var host = NewHost(hostId, isPrimary: true);
await host.StartAsync(ct);
host.IsClaimed.ShouldBeTrue("the fixture broker accepted the claim connection");
// Subscribe only NOW — after the birth was published — so only a retained message can satisfy it.
await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct);
var body = await watcher.WaitForOneAsync(Timeout, ct);
SparkplugStatePayload.TryDecode(body, out var online, out var timestamp).ShouldBeTrue();
online.ShouldBeTrue();
timestamp.ShouldBeGreaterThan(0);
await host.StopAsync(ct);
await ClearRetainedAsync(hostId, ct);
}
[Fact]
public async Task An_abrupt_death_leaves_a_retained_OFFLINE_because_the_will_fires()
{
// THE leg the whole design exists for, and the one no unit test can reach: proof that the
// broker is holding our death certificate and delivers it when the socket dies without a
// goodbye. Without it a crashed node would leave a retained ONLINE asserting forever that a
// dead host is alive — strictly worse than never publishing at all.
SkipIfUnconfigured();
var hostId = NewHostId();
var ct = TestContext.Current.CancellationToken;
var host = NewHost(hostId, isPrimary: true);
await host.StartAsync(ct);
host.IsClaimed.ShouldBeTrue();
await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct);
var birth = await watcher.WaitForOneAsync(Timeout, ct);
SparkplugStatePayload.TryDecode(birth, out var wasOnline, out var birthTimestamp).ShouldBeTrue();
wasOnline.ShouldBeTrue();
// Kill the socket with no DISCONNECT. A clean teardown would make the broker DISCARD the Will
// by protocol rule, so this test would then be observing the explicit stand-down publish and
// would pass against an implementation that never armed a Will.
await host.AbandonClaimForTests();
var death = await watcher.WaitForMatchAsync(
p => SparkplugStatePayload.TryDecode(p, out var o, out _) && !o, Timeout, ct);
SparkplugStatePayload.TryDecode(death, out _, out var deathTimestamp).ShouldBeTrue();
// Sparkplug 3.0 pairs a host's birth and death by timestamp — an edge node may ignore an
// OFFLINE older than the ONLINE it holds. Mismatched timestamps produce two well-formed
// messages and a death certificate conformant hosts are entitled to throw away.
deathTimestamp.ShouldBe(birthTimestamp);
await host.DisposeAsync();
await ClearRetainedAsync(hostId, ct);
}
[Fact]
public async Task A_clean_shutdown_leaves_a_retained_OFFLINE_without_waiting_for_the_will()
{
SkipIfUnconfigured();
var hostId = NewHostId();
var ct = TestContext.Current.CancellationToken;
var host = NewHost(hostId, isPrimary: true);
await host.StartAsync(ct);
await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct);
await watcher.WaitForMatchAsync(
p => SparkplugStatePayload.TryDecode(p, out var o, out _) && o, Timeout, ct);
await host.StopAsync(ct);
// Promptly — the assertion is that this arrives well inside the keep-alive window it would take
// the broker to notice a dead socket, which is why the explicit publish exists at all.
await watcher.WaitForMatchAsync(
p => SparkplugStatePayload.TryDecode(p, out var o, out _) && !o, Timeout, ct);
await host.DisposeAsync();
await ClearRetainedAsync(hostId, ct);
}
[Fact]
public async Task A_node_that_is_not_primary_publishes_nothing_at_all()
{
// The pair-node answer. Not merely "does not win the race" — a Secondary must never open a
// claim connection, because the Will it would register is the hazard: its own death would then
// publish OFFLINE over a perfectly healthy Primary's ONLINE and stop the whole group.
SkipIfUnconfigured();
var hostId = NewHostId();
var ct = TestContext.Current.CancellationToken;
await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct);
await using var host = NewHost(hostId, isPrimary: false);
await host.StartAsync(ct);
host.ClaimAttemptsForTests.ShouldBe(0);
// Give the reconcile loop several ticks to misbehave in.
await Task.Delay(TimeSpan.FromSeconds(3), ct);
host.IsClaimed.ShouldBeFalse();
host.ClaimAttemptsForTests.ShouldBe(0);
watcher.Count.ShouldBe(0, "a Secondary must not touch the STATE topic");
}
[Fact]
public async Task A_failover_moves_the_claim_offline_then_online_again()
{
// Both halves of the accepted trade, observed: losing Primary publishes OFFLINE (a genuine
// state transition — for a window there really is no host), and regaining it republishes
// ONLINE. The republish is the half that would be easy to omit; without it a node that failed
// back would run as Primary while every edge node believed it was gone.
SkipIfUnconfigured();
var hostId = NewHostId();
var ct = TestContext.Current.CancellationToken;
var role = new MutableRole(isPrimary: true);
await using var host = new SparkplugPrimaryHost(Options(hostId), "live-failover", role);
await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct);
await host.StartAsync(ct);
await watcher.WaitForMatchAsync(
p => SparkplugStatePayload.TryDecode(p, out var o, out _) && o, Timeout, ct);
role.IsPrimary = false;
await watcher.WaitForMatchAsync(
p => SparkplugStatePayload.TryDecode(p, out var o, out _) && !o, Timeout, ct);
await WaitUntilAsync(() => !host.IsClaimed, ct);
var offlineSeen = watcher.CountMatching(
p => SparkplugStatePayload.TryDecode(p, out var o, out _) && !o);
role.IsPrimary = true;
await WaitUntilAsync(() => host.IsClaimed, ct);
// A SECOND online, after the offline — not merely "an online exists", which the birth already
// satisfied before the failover even started.
await WaitUntilAsync(
() => watcher.CountMatching(p => SparkplugStatePayload.TryDecode(p, out var o, out _) && o) >= 2,
ct);
offlineSeen.ShouldBeGreaterThanOrEqualTo(1);
await host.StopAsync(ct);
await ClearRetainedAsync(hostId, ct);
}
// ---- helpers -----------------------------------------------------------------------------------
/// <summary>A per-test host id: the retained STATE outlives the test that wrote it.</summary>
private static string NewHostId() => $"otopcua-live-{Guid.NewGuid():N}"[..28];
private MqttDriverOptions Options(string hostId) => new()
{
Host = fixture.PlainHost,
Port = fixture.PlainPort,
Username = fixture.Username,
Password = fixture.Password,
UseTls = false,
ClientId = $"otopcua-live-{Guid.NewGuid():N}"[..20],
ConnectTimeoutSeconds = 10,
Mode = MqttMode.SparkplugB,
Sparkplug = new MqttSparkplugOptions
{
GroupId = "OtOpcUaSim",
HostId = hostId,
ActAsPrimaryHost = true,
},
};
private SparkplugPrimaryHost NewHost(string hostId, bool isPrimary) =>
new(Options(hostId), $"live-{hostId}", new MutableRole(isPrimary));
/// <summary>
/// Clears the retained STATE this test left behind — a zero-length retained publish is MQTT's
/// "forget this". Leaving a retained ONLINE for a host that no longer exists is precisely the
/// lie the feature prevents.
/// </summary>
private async Task ClearRetainedAsync(string hostId, CancellationToken ct)
{
var client = new MqttClientFactory().CreateMqttClient();
try
{
await client.ConnectAsync(
new MqttClientOptionsBuilder()
.WithTcpServer(fixture.PlainHost, fixture.PlainPort)
.WithCredentials(fixture.Username, fixture.Password)
.WithClientId($"otopcua-live-clean-{Guid.NewGuid():N}"[..23])
.WithTimeout(TimeSpan.FromSeconds(10))
.Build(),
ct);
await client.PublishAsync(
new MqttApplicationMessageBuilder()
.WithTopic(SparkplugTopic.FormatState(hostId))
.WithPayload(Array.Empty<byte>())
.WithRetainFlag(true)
.Build(),
ct);
await client.DisconnectAsync(cancellationToken: ct);
}
finally
{
client.Dispose();
}
}
private static async Task WaitUntilAsync(Func<bool> condition, CancellationToken ct)
{
var deadline = DateTime.UtcNow + Timeout;
while (!condition())
{
if (DateTime.UtcNow >= deadline)
{
throw new TimeoutException($"Condition was not satisfied within {Timeout}.");
}
await Task.Delay(100, ct);
}
}
/// <summary>A role source a test can flip, standing in for a redundancy failover.</summary>
private sealed class MutableRole(bool isPrimary) : IDriverPrimaryRoleSource
{
private volatile bool _isPrimary = isPrimary;
public bool IsPrimary
{
get => _isPrimary;
set => _isPrimary = value;
}
public bool ShouldActAsPrimary => _isPrimary;
public void PublishPrimaryRole(bool shouldActAsPrimary) => _isPrimary = shouldActAsPrimary;
}
/// <summary>Subscribes one host's STATE topic and records every body seen, in order.</summary>
private sealed class StateWatcher : IAsyncDisposable
{
private readonly IMqttClient _client;
private readonly ConcurrentQueue<byte[]> _bodies = new();
private StateWatcher(IMqttClient client) => _client = client;
public int Count => _bodies.Count;
public static async Task<StateWatcher> StartAsync(MqttFixture fixture, string hostId, CancellationToken ct)
{
var client = new MqttClientFactory().CreateMqttClient();
var watcher = new StateWatcher(client);
client.ApplicationMessageReceivedAsync += args =>
{
watcher._bodies.Enqueue(args.ApplicationMessage.Payload.ToArray());
return Task.CompletedTask;
};
await client.ConnectAsync(
new MqttClientOptionsBuilder()
.WithTcpServer(fixture.PlainHost, fixture.PlainPort)
.WithCredentials(fixture.Username, fixture.Password)
.WithClientId($"otopcua-live-state-{Guid.NewGuid():N}"[..23])
.WithTimeout(TimeSpan.FromSeconds(10))
.Build(),
ct);
await client.SubscribeAsync(
new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(f => f.WithTopic(SparkplugTopic.FormatState(hostId)))
.Build(),
ct);
return watcher;
}
public int CountMatching(Func<byte[], bool> predicate) => _bodies.Count(b => Safe(predicate, b));
public async Task<byte[]> WaitForOneAsync(TimeSpan timeout, CancellationToken ct)
=> await WaitForMatchAsync(_ => true, timeout, ct);
public async Task<byte[]> WaitForMatchAsync(Func<byte[], bool> predicate, TimeSpan timeout, CancellationToken ct)
{
var deadline = DateTime.UtcNow + timeout;
while (true)
{
if (_bodies.FirstOrDefault(b => Safe(predicate, b)) is { } match)
{
return match;
}
if (DateTime.UtcNow >= deadline)
{
throw new TimeoutException(
$"No matching STATE message within {timeout} ({_bodies.Count} seen, none matching).");
}
await Task.Delay(100, ct);
}
}
private static bool Safe(Func<byte[], bool> predicate, byte[] body)
{
try { return predicate(body); }
catch { return false; }
}
public async ValueTask DisposeAsync()
{
try
{
if (_client.IsConnected)
{
await _client.DisconnectAsync();
}
}
catch
{
// Teardown only.
}
finally
{
_client.Dispose();
}
}
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs
+263
View File
@@ -0,0 +1,263 @@
using System.Text;
using MQTTnet;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
/// <summary>
/// Unit tests for the Sparkplug primary-host claim (Gitea #511) — the parts that are decidable
/// without a broker: the STATE bodies, the Will registration, and the gate that decides whether a
/// claim is attempted at all.
/// </summary>
/// <remarks>
/// The claim/release state machine itself needs a broker and is covered by
/// <c>MqttPrimaryHostLiveTests</c> in the integration suite. What is pinned here is everything a
/// live test could not tell apart from a working implementation by accident — above all that the
/// birth and the Will carry the SAME timestamp, which a live test would see as two well-formed
/// messages.
/// </remarks>
[Trait("Category", "Unit")]
public sealed class SparkplugPrimaryHostTests
{
private static MqttDriverOptions Options(
bool actAsPrimaryHost = true,
string hostId = "otopcua-primary",
MqttMode mode = MqttMode.SparkplugB)
=> new()
{
Host = "broker.example",
Port = 1883,
ClientId = "otopcua",
Mode = mode,
Sparkplug = new MqttSparkplugOptions
{
GroupId = "OtOpcUaSim",
HostId = hostId,
ActAsPrimaryHost = actAsPrimaryHost,
},
};
// ---- The matched-timestamp invariant --------------------------------------------------------
[Fact]
public void The_birth_and_the_will_carry_the_same_timestamp()
{
// Sparkplug 3.0 pairs a host's birth and death certificates by their timestamp, so an edge node
// may ignore an OFFLINE older than the ONLINE it holds. The two bodies are necessarily built at
// different moments — the Will goes in the CONNECT packet, the birth is published after CONNACK
// — so reading the clock twice is the natural mistake, and it produces two individually
// well-formed messages that no live test would flag.
var session = SparkplugStatePayload.ForSession("otopcua-primary", 1_770_000_000_123);
SparkplugStatePayload.TryDecode(session.OnlineBody, out var online, out var birthTs).ShouldBeTrue();
SparkplugStatePayload.TryDecode(session.OfflineBody, out var offline, out var deathTs).ShouldBeTrue();
online.ShouldBeTrue();
offline.ShouldBeFalse();
birthTs.ShouldBe(1_770_000_000_123);
deathTs.ShouldBe(birthTs);
}
[Fact]
public void The_state_topic_is_the_v3_form()
=> SparkplugStatePayload.ForSession("scada-primary", 1).Topic.ShouldBe("spBv1.0/STATE/scada-primary");
[Fact]
public void The_timestamp_is_a_bare_json_number()
{
// A quoted timestamp still parses as a valid JSON object, so a serializer choosing string
// formatting for a long would be silently wrong to every conformant host.
var body = Encoding.UTF8.GetString(SparkplugStatePayload.Encode(online: true, 1_770_000_000_123));
body.ShouldBe("{\"online\":true,\"timestamp\":1770000000123}");
}
[Fact]
public void The_body_survives_a_non_invariant_culture()
{
var previous = Thread.CurrentThread.CurrentCulture;
Thread.CurrentThread.CurrentCulture = new System.Globalization.CultureInfo("de-DE");
try
{
Encoding.UTF8.GetString(SparkplugStatePayload.Encode(online: false, 1_770_000_000_123))
.ShouldBe("{\"online\":false,\"timestamp\":1770000000123}");
}
finally
{
Thread.CurrentThread.CurrentCulture = previous;
}
}
// ---- The Will actually reaches the CONNECT packet --------------------------------------------
[Fact]
public void The_will_is_registered_on_the_client_options_retained_at_qos1()
{
// The single most important assertion in this file. A retained ONLINE with no Will behind it is
// strictly worse than publishing nothing: a crashed node then asserts forever that a dead host
// is alive. This is the check that the death certificate is armed at CONNECT rather than merely
// constructed.
var session = SparkplugStatePayload.ForSession("otopcua-primary", 42);
var will = new MqttLastWill(session.Topic, session.OfflineBody, SparkplugStatePayload.StateQos, Retain: true);
var built = MqttClientOptionsFactory.BuildClientOptions(
Options(), SparkplugPrimaryHost.ClientIdSuffix, logger: null, lastWill: will, cleanSessionOverride: true);
built.WillTopic.ShouldBe("spBv1.0/STATE/otopcua-primary");
built.WillRetain.ShouldBeTrue();
built.WillQualityOfServiceLevel.ShouldBe(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);
SparkplugStatePayload.TryDecode(built.WillPayload, out var online, out _).ShouldBeTrue();
online.ShouldBeFalse();
}
[Fact]
public void No_will_is_registered_when_none_is_supplied()
// Every other connection this repo opens — ingest, browse, probe — must stay Will-free. A Will
// leaking onto the ingest session would publish a host death when ingest merely reconnects.
=> MqttClientOptionsFactory.BuildClientOptions(Options(), clientIdSuffix: null).WillTopic.ShouldBeNull();
[Fact]
public void The_claim_connection_forces_clean_session_over_the_configured_value()
{
// A broker resuming a previous session for this client id may keep the OLD session's Will and
// discard the one this CONNECT carries — a claim with no death certificate behind it.
var persistent = Options() with { CleanSession = false };
MqttClientOptionsFactory.BuildClientOptions(persistent, "-spbhost", cleanSessionOverride: true)
.CleanSession.ShouldBeTrue();
// ...and the override is genuinely an override, not a constant.
MqttClientOptionsFactory.BuildClientOptions(persistent, clientIdSuffix: null)
.CleanSession.ShouldBeFalse();
}
[Fact]
public void The_claim_connection_uses_a_distinct_client_id()
{
// Two MQTT clients presenting one client id is not an error the broker reports: it evicts the
// older one. Ingest and the claim would take turns killing each other, forever, looking like a
// flapping broker.
var ingest = MqttClientOptionsFactory.BuildClientOptions(Options(), clientIdSuffix: null).ClientId;
var claim = MqttClientOptionsFactory
.BuildClientOptions(Options(), SparkplugPrimaryHost.ClientIdSuffix).ClientId;
claim.ShouldNotBe(ingest);
claim.ShouldBe("otopcua-spbhost");
}
// ---- Whether a claim is attempted at all -----------------------------------------------------
[Fact]
public void A_configured_claim_is_claimable()
=> SparkplugPrimaryHost.CanClaim(Options()).ShouldBeTrue();
[Theory]
[InlineData("")]
[InlineData(" ")]
public void A_blank_host_id_is_refused_rather_than_publishing_to_a_malformed_topic(string hostId)
{
// "spBv1.0/STATE/" is a legal MQTT topic, so a publish would succeed and reach nobody.
SparkplugPrimaryHost.CanClaim(Options(hostId: hostId)).ShouldBeFalse();
SparkplugPrimaryHost.DescribeUnusable(Options(hostId: hostId)).ShouldNotBeNullOrWhiteSpace();
}
[Fact]
public void The_flag_is_ignored_outside_sparkplug_mode_and_says_so()
{
var plain = Options(mode: MqttMode.Plain);
SparkplugPrimaryHost.CanClaim(plain).ShouldBeFalse();
SparkplugPrimaryHost.DescribeUnusable(plain).ShouldNotBeNullOrWhiteSpace();
}
[Fact]
public void An_unset_flag_produces_no_complaint()
{
// DescribeUnusable must stay silent about a configuration that never asked for anything, or the
// driver warns on every ordinary Sparkplug deployment.
SparkplugPrimaryHost.CanClaim(Options(actAsPrimaryHost: false)).ShouldBeFalse();
SparkplugPrimaryHost.DescribeUnusable(Options(actAsPrimaryHost: false)).ShouldBeNull();
SparkplugPrimaryHost.DescribeUnusable(Options(actAsPrimaryHost: false, hostId: "")).ShouldBeNull();
}
[Fact]
public void Constructing_a_claim_without_a_host_id_throws_rather_than_half_working()
=> Should.Throw<ArgumentException>(() =>
new SparkplugPrimaryHost(Options(hostId: " "), "drv-1", UnmanagedDriverPrimaryRoleSource.Instance));
// ---- The gate ---------------------------------------------------------------------------------
[Fact]
public void A_driver_with_no_role_source_claims()
{
// The documented fallback for a standalone host or a test: with nothing publishing a role there
// is no peer to collide with, and defaulting closed would make the feature silently inert on the
// single-node deployment that is its commonest use.
UnmanagedDriverPrimaryRoleSource.Instance.ShouldActAsPrimary.ShouldBeTrue();
// Publishing into it is a no-op rather than a throw, so a caller need not special-case it.
UnmanagedDriverPrimaryRoleSource.Instance.PublishPrimaryRole(false);
UnmanagedDriverPrimaryRoleSource.Instance.ShouldActAsPrimary.ShouldBeTrue();
}
[Fact]
public async Task A_secondary_never_even_attempts_a_claim_connection()
{
// The whole point of gating. Asserted on ATTEMPTS, not on IsClaimed: both nodes point at the
// same unreachable broker here, so IsClaimed is false either way and an IsClaimed-only
// assertion would pass just as happily with the gate deleted. See ClaimAttemptsForTests.
var host = new SparkplugPrimaryHost(
Unreachable(), "drv-1", new FixedRole(shouldActAsPrimary: false));
await host.StartAsync(TestContext.Current.CancellationToken);
host.ClaimAttemptsForTests.ShouldBe(0);
host.IsClaimed.ShouldBeFalse();
await host.DisposeAsync();
}
[Fact]
public async Task A_primary_attempts_the_claim_and_survives_a_broker_it_cannot_reach()
{
// The falsifying control for the test above — same options, same broker, opposite verdict — so
// "0 attempts" there is a fact about the gate rather than about the fixture.
//
// It also pins the swallow: the claim is advisory and ingest is the data plane, so a broker that
// refuses it must cost a warning, not the driver. And IsClaimed must stay honest — a connection
// whose ONLINE never landed is not a claim.
var host = new SparkplugPrimaryHost(Unreachable(), "drv-1", new FixedRole(shouldActAsPrimary: true));
await Should.NotThrowAsync(() => host.StartAsync(TestContext.Current.CancellationToken));
host.ClaimAttemptsForTests.ShouldBe(1);
host.IsClaimed.ShouldBeFalse();
await host.DisposeAsync();
}
/// <summary>A broker that refuses instantly, so a connect attempt is observable but not slow.</summary>
private static MqttDriverOptions Unreachable() =>
Options() with { Host = "127.0.0.1", Port = 1, ConnectTimeoutSeconds = 1 };
[Fact]
public async Task Stopping_an_unclaimed_host_is_a_no_op()
{
var host = new SparkplugPrimaryHost(Options(), "drv-1", new FixedRole(shouldActAsPrimary: false));
await Should.NotThrowAsync(() => host.StopAsync(TestContext.Current.CancellationToken));
// Twice: teardown runs from both ShutdownAsync and DisposeAsync, so a second release that threw
// — or that published a second OFFLINE — would surface on every ordinary driver shutdown.
await Should.NotThrowAsync(async () => await host.DisposeAsync());
await Should.NotThrowAsync(async () => await host.DisposeAsync());
}
private sealed class FixedRole(bool shouldActAsPrimary) : IDriverPrimaryRoleSource
{
public bool ShouldActAsPrimary { get; private set; } = shouldActAsPrimary;
public void PublishPrimaryRole(bool value) => ShouldActAsPrimary = value;
}
}