diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md
index d13bbf16..f8a6f5a4 100644
--- a/docs/drivers/Mqtt.md
+++ b/docs/drivers/Mqtt.md
@@ -325,6 +325,9 @@ and this driver ships all three or none of it
2. a matching **retained `OFFLINE` armed as the MQTT Last Will**, so a crash still reports the truth;
3. an explicit `OFFLINE` on a clean stand-down, so edge nodes do not wait out a keep-alive timeout.
+A claim the broker keeps refusing backs off (STAB-27): 2 s, 4 s, 8 s … capped at 30 s between attempts, one
+Warning per attempt, reset by a successful claim or a stand-down. The role is still sampled every second.
+
Set `Sparkplug.HostId` and `Sparkplug.ActAsPrimaryHost: true`. A blank host id is **refused** with a
warning — the driver does not publish to `spBv1.0/STATE/`, which is a legal topic reaching nobody.
@@ -460,8 +463,12 @@ every MQTT node materialized read-only regardless of what the operator authored.
| **Plain** | the tag's `writeTopic` | the value as UTF-8 text (`Raw`/`Scalar`) or a bare JSON literal (`Json`) |
| **Sparkplug B** | `spBv1.0/{group}/DCMD/{node}/{device}` for a device metric, `spBv1.0/{group}/NCMD/{node}` for an edge-node metric | a Sparkplug `Payload` carrying the metric **by name**, with an explicit `Datatype` and no `seq` |
-Both publish at **QoS 0, `retain: false`**. Retain is the one that matters and is deliberately not
-configurable: a retained command would be replayed by the broker to the device on *every* future
+Both publish with `retain: false`. A **Plain** write goes at **QoS 1** (STAB-26): the publish completes only
+once the broker PUBACKs it, so `Good` means "the broker accepted it", and a broker refusal or a connection
+that died under the send is `Bad`. **Sparkplug commands stay at QoS 0** — a redelivered NCMD/DCMD would be
+re-applied. A Sparkplug metric the tag did not give a `dataType` is commanded with the type its live birth
+certificate declared (STAB-25), the same authored-beats-birth precedence reads use. Retain is the one that
+matters and is deliberately not configurable: a retained command would be replayed by the broker to the device on *every* future
subscribe — including the device's own reconnect — so one stale write would be re-applied forever.
### `writeTopic` is separate from `topic`, and its absence is meaningful
diff --git a/docs/drivers/Sql.md b/docs/drivers/Sql.md
index 37803f39..a31617ca 100644
--- a/docs/drivers/Sql.md
+++ b/docs/drivers/Sql.md
@@ -92,7 +92,10 @@ Both value-bearing fields — `whereValue` here and `keyValue` on the `KeyValue`
authored as a JSON string, number or boolean, and are captured as text. They are **bound as
parameters** either way, and matched back as text (the key cell is stringified), so a key column
that holds an INTEGER station id can be authored either `"7"` or `7`. A JSON object, array or `null`
-is not a value and is refused. (They were read asymmetrically until 2026-08-10, so a numerically
+is not a value and is refused. On **PostgreSQL** a string bind is sent untyped (`NpgsqlDbType.Unknown`,
+`PostgresDialect.ConfigureParameter`, STAB-24) so the server infers its type from the compared column;
+bound as `text`, a `"7"` against an `integer` key fails with `operator does not exist: integer = text`.
+(They were read asymmetrically until 2026-08-10, so a numerically
authored `keyValue` refused the tag while the identically-bound `whereValue` was served — and the
deploy gate reported the key as *missing* when it was right there in the blob.)
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
index d7f89c77..63bce6d1 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
@@ -229,7 +229,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
{
state.ProbeCts = new CancellationTokenSource();
var ct = state.ProbeCts.Token;
- _ = Task.Run(() => ProbeLoopAsync(state, ct), ct);
+ state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct);
}
}
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
@@ -245,10 +245,13 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
// handles alive. Mirrors the body of ShutdownAsync without awaiting the poll engine
// (nothing has been subscribed yet at init time).
foreach (var state in _devices.Values)
- {
try { state.ProbeCts?.Cancel(); } catch { }
+ JoinProbeLoopsSync();
+ foreach (var state in _devices.Values)
+ {
state.ProbeCts?.Dispose();
state.ProbeCts = null;
+ state.ProbeTask = null;
state.DisposeRuntimes();
}
_devices.Clear();
@@ -266,15 +269,45 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
+ /// Upper bound on how long shutdown waits for the probe loops to observe cancellation.
+ internal static readonly TimeSpan LoopJoinTimeout = TimeSpan.FromSeconds(5);
+
+ ///
+ /// Synchronous bounded join of the (already-cancelled) probe loops for the sync teardown paths.
+ /// The loop body uses ConfigureAwait(false) throughout, so blocking here cannot deadlock.
+ ///
+ private void JoinProbeLoopsSync()
+ {
+ var loops = _devices.Values.Select(d => d.ProbeTask).OfType().ToArray();
+ if (loops.Length == 0) return;
+ try { Task.WaitAll(loops, LoopJoinTimeout); }
+ catch (AggregateException) { /* loops observe cancellation; a fault is not actionable at teardown */ }
+ }
+
///
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
await _poll.DisposeAsync().ConfigureAwait(false);
+ // STAB-6: cancel the probe loops, JOIN them (bounded), then dispose the CTSs and runtimes.
+ // The loop used to be fire-and-forget, so one parked in a probe-runtime create/read could
+ // outlive shutdown and keep a libplctag handle alive past DisposeRuntimes.
+ foreach (var state in _devices.Values)
+ try { state.ProbeCts?.Cancel(); } catch { }
+ var loops = _devices.Values.Select(d => d.ProbeTask).OfType().ToList();
+ if (loops.Count > 0)
+ {
+ var all = Task.WhenAll(loops);
+ var finished = await Task.WhenAny(all, Task.Delay(LoopJoinTimeout, CancellationToken.None)).ConfigureAwait(false);
+ if (finished != all)
+ _logger.LogWarning(
+ "AbLegacy shutdown: {Count} probe loop(s) did not stop within {Timeout}; proceeding with teardown. Driver={DriverInstanceId}",
+ loops.Count, LoopJoinTimeout, _driverInstanceId);
+ }
foreach (var state in _devices.Values)
{
- try { state.ProbeCts?.Cancel(); } catch { }
state.ProbeCts?.Dispose();
state.ProbeCts = null;
+ state.ProbeTask = null;
state.DisposeRuntimes();
}
_devices.Clear();
@@ -908,10 +941,13 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
// can't poison the rest of the disposal chain.
try { _poll.DisposeAsync().ConfigureAwait(false).GetAwaiter().GetResult(); } catch { }
foreach (var state in _devices.Values)
- {
try { state.ProbeCts?.Cancel(); } catch { }
+ JoinProbeLoopsSync();
+ foreach (var state in _devices.Values)
+ {
state.ProbeCts?.Dispose();
state.ProbeCts = null;
+ state.ProbeTask = null;
state.DisposeRuntimes();
}
_devices.Clear();
@@ -1037,6 +1073,11 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
///
public CancellationTokenSource? ProbeCts { get; set; }
+ ///
+ /// Gets or sets the probe loop task, retained so shutdown can join it (STAB-6).
+ ///
+ public Task? ProbeTask { get; set; }
+
///
/// Gets or sets a value indicating whether the probe has been initialized.
///
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs
index 9fb93432..07fc913c 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs
@@ -277,7 +277,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
{
state.ProbeCts = new CancellationTokenSource();
var ct = state.ProbeCts.Token;
- _ = Task.Run(() => ProbeLoopAsync(state, ct), ct);
+ state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct);
}
}
@@ -287,7 +287,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
{
state.RecycleCts = new CancellationTokenSource();
var ct = state.RecycleCts.Token;
- _ = Task.Run(() => RecycleLoopAsync(state, ct), ct);
+ state.RecycleTask = Task.Run(() => RecycleLoopAsync(state, ct), ct);
}
}
@@ -305,7 +305,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
{
state.FixedTreeCts = new CancellationTokenSource();
var ct = state.FixedTreeCts.Token;
- _ = Task.Run(() => FixedTreeLoopAsync(state, ct), ct);
+ state.FixedTreeTask = Task.Run(() => FixedTreeLoopAsync(state, ct), ct);
}
}
@@ -326,6 +326,9 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
+ /// Upper bound on how long shutdown waits for the background loops to observe cancellation.
+ internal static readonly TimeSpan LoopJoinTimeout = TimeSpan.FromSeconds(5);
+
///
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
@@ -335,22 +338,43 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
await proj.DisposeAsync().ConfigureAwait(false);
_alarmProjection = null;
}
+ // STAB-23: cancel every loop first, JOIN them (bounded), and only then dispose the CTSs and the
+ // client. Fire-and-forget loops used to outlive shutdown: one parked in EnsureConnectedAsync could
+ // finish its connect AFTER DisposeClient and leave a live socket on a shut-down driver.
+ var loops = new List();
foreach (var state in _devices.Values)
{
// Cancel-then-dispose can race in tight shutdown loops; swallowing is intentional
// but we now log the cause so a noisy shutdown leaves a Debug trace.
try { state.ProbeCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling probe CTS for {Host} failed", state.Options.HostAddress); }
- state.ProbeCts?.Dispose();
- state.ProbeCts = null;
try { state.RecycleCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling recycle CTS for {Host} failed", state.Options.HostAddress); }
- state.RecycleCts?.Dispose();
- state.RecycleCts = null;
try { state.FixedTreeCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling fixed-tree CTS for {Host} failed", state.Options.HostAddress); }
+ if (state.ProbeTask is { } pt) loops.Add(pt);
+ if (state.RecycleTask is { } rt) loops.Add(rt);
+ if (state.FixedTreeTask is { } ft) loops.Add(ft);
+ }
+ if (loops.Count > 0)
+ {
+ var all = Task.WhenAll(loops);
+ var finished = await Task.WhenAny(all, Task.Delay(LoopJoinTimeout, CancellationToken.None)).ConfigureAwait(false);
+ if (finished != all)
+ _logger.LogWarning("FOCAS shutdown: {Count} background loop(s) did not stop within {Timeout}; proceeding with teardown",
+ loops.Count, LoopJoinTimeout);
+ }
+ foreach (var state in _devices.Values)
+ {
+ state.ProbeCts?.Dispose();
+ state.ProbeCts = null;
+ state.ProbeTask = null;
+ state.RecycleCts?.Dispose();
+ state.RecycleCts = null;
+ state.RecycleTask = null;
state.FixedTreeCts?.Dispose();
state.FixedTreeCts = null;
+ state.FixedTreeTask = null;
state.DisposeClient();
}
_devices.Clear();
@@ -1575,6 +1599,12 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
public CancellationTokenSource? RecycleCts { get; set; }
/// Gets or sets the cancellation token source for the fixed-tree loop.
public CancellationTokenSource? FixedTreeCts { get; set; }
+ /// The probe loop task, retained so shutdown can join it (STAB-23).
+ public Task? ProbeTask { get; set; }
+ /// The recycle loop task, retained so shutdown can join it (STAB-23).
+ public Task? RecycleTask { get; set; }
+ /// The fixed-tree loop task, retained so shutdown can join it (STAB-23).
+ public Task? FixedTreeTask { get; set; }
/// Gets or sets the fixed-tree cache for this device.
public FocasFixedTreeCache? FixedTreeCache { get; set; }
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
index 5001c1aa..3da7bf61 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
@@ -327,6 +327,20 @@ public sealed class MTConnectDriver
/// The shared /sample pump task. Guarded by .
private Task? _pumpTask;
+ ///
+ /// The the running pump was started against (STAB-20), guarded by
+ /// like the rest of the control block. A pump whose generation is
+ /// not the current one is bound to a client a lifecycle restart has retired.
+ ///
+ private long _pumpGeneration;
+
+ ///
+ /// Bumped each time installs a NEW (not on a
+ /// with-copy that only advances the cursor or the probe cache). Written under
+ /// .
+ ///
+ private long _sessionGeneration;
+
///
/// The Agent answered /sample with something that cannot be streamed at all, so the
/// pump gave up. Latched — reconnecting reproduces the identical answer forever, and a
@@ -1808,7 +1822,23 @@ public sealed class MTConnectDriver
// revived by a resubscribe. The genuinely-unrepeatable case is the latch below, not this.
if (_pumpTask is { IsCompleted: false })
{
- return;
+ if (_pumpGeneration == _sessionGeneration)
+ {
+ return;
+ }
+
+ // STAB-20: a Subscribe can land between StopSampleStreamAsync clearing the registry slots and
+ // the teardown that retires the client — StopSampleStreamAsync is still awaiting the OLD pump
+ // then — and starts a pump on the session about to be disposed. Treating that pump as "the
+ // running stream" would leave the freshly-installed session with no live stream at all
+ // (the stale pump just reads its disposed client as a dropped connection). Retire it: it is
+ // cancelled and can only fail against its disposed client.
+ try { _pumpCts?.Cancel(); }
+ catch (ObjectDisposedException) { /* already going away */ }
+
+ _logger.LogInformation(
+ "MTConnect driver {DriverInstanceId} replaced a /sample pump bound to a retired Agent session.",
+ _driverInstanceId);
}
if (_streamUnsupported)
@@ -1856,6 +1886,7 @@ public sealed class MTConnectDriver
var options = Volatile.Read(ref _options);
var cts = new CancellationTokenSource();
_pumpCts = cts;
+ _pumpGeneration = _sessionGeneration;
_pumpTask = Task.Run(
() => RunSampleStreamAsync(
session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, cts.Token),
@@ -2695,6 +2726,11 @@ public sealed class MTConnectDriver
// One publish, so a concurrent lock-free reader sees the new client, the new probe
// cache, and the pump's opening cursor together or none of them.
+ lock (_subscriptionLock)
+ {
+ _sessionGeneration++;
+ }
+
Volatile.Write(
ref _session,
new AgentSession(client, probe, current.InstanceId, current.NextSequence));
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs
index 6d0b1125..5ebd5d96 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs
@@ -74,9 +74,20 @@ public enum MqttWriteEncodeError
///
public static class MqttWriteEncoder
{
- /// QoS for an outbound write. See the remarks on .
+ ///
+ /// QoS for a Sparkplug NCMD/DCMD command. See the remarks on . Plain-mode
+ /// writes use instead.
+ ///
public const int WriteQos = 0;
+ ///
+ /// QoS for a Plain-mode writeTopic publish (STAB-26). At QoS 1 the publish completes only
+ /// once the broker has PUBACKed (or refused) it, so a Good write status means "the broker accepted
+ /// the message" rather than "the bytes left this socket"; at QoS 0 a broker-side refusal or a
+ /// connection that dropped under the send was indistinguishable from success.
+ ///
+ public const int PlainWriteQos = 1;
+
///
/// Encodes one write for the tag under the driver's .
///
@@ -84,9 +95,16 @@ public static class MqttWriteEncoder
/// The resolved tag definition.
/// The value to write.
/// The topic + payload to publish, when this returns .
+ ///
+ /// Sparkplug only (STAB-25): the type to encode the metric as when the tag did not author one — the
+ /// type the metric's live birth certificate declared, the same authored ? authored : birth
+ /// precedence the ingest path coerces reads with. encodes as
+ /// , which for an unauthored tag is only the record's
+ /// default and would command the metric with the wrong wire type. Ignored in Plain mode.
+ ///
/// The encode outcome.
///
- /// QoS 0, retain . Retain is the one that matters: a retained
+ /// Retain always; QoS 1 for Plain, 0 for Sparkplug commands. Retain is the one that matters: a retained
/// command would be replayed by the broker to the device on every future subscribe — including
/// the device's own reconnect — so a single stale write would be re-applied forever. This is the
/// same hazard avoids for NCMD, and the reason neither is
@@ -96,13 +114,14 @@ public static class MqttWriteEncoder
MqttMode mode,
MqttTagDefinition def,
object? value,
- out MqttEncodedWrite encoded)
+ out MqttEncodedWrite encoded,
+ DriverDataType? effectiveDataType = null)
{
ArgumentNullException.ThrowIfNull(def);
encoded = default;
return mode == MqttMode.SparkplugB
- ? TryEncodeSparkplug(def, value, out encoded)
+ ? TryEncodeSparkplug(def, value, effectiveDataType, out encoded)
: TryEncodePlain(def, value, out encoded);
}
@@ -163,7 +182,7 @@ public static class MqttWriteEncoder
: value.ToString() ?? "";
}
- encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text));
+ encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text), PlainWriteQos);
return MqttWriteEncodeError.None;
}
@@ -175,6 +194,7 @@ public static class MqttWriteEncoder
private static MqttWriteEncodeError TryEncodeSparkplug(
MqttTagDefinition def,
object? value,
+ DriverDataType? effectiveDataType,
out MqttEncodedWrite encoded)
{
encoded = default;
@@ -206,7 +226,7 @@ public static class MqttWriteEncoder
}
var metric = new Payload.Types.Metric { Name = metricName };
- var error = ApplyValue(metric, def.DataType, value);
+ var error = ApplyValue(metric, def.DataTypeAuthored ? def.DataType : effectiveDataType ?? def.DataType, value);
if (error != MqttWriteEncodeError.None)
{
return error;
@@ -227,7 +247,7 @@ public static class MqttWriteEncoder
edgeNodeId,
isDevice ? def.DeviceId : null);
- encoded = new MqttEncodedWrite(topic, payload.ToByteArray());
+ encoded = new MqttEncodedWrite(topic, payload.ToByteArray(), WriteQos);
return MqttWriteEncodeError.None;
}
@@ -344,4 +364,5 @@ public static class MqttWriteEncoder
/// One encoded outbound write: where it goes and what it carries.
/// The concrete topic to publish on.
/// The encoded message body.
-public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload);
+/// The MQTT QoS to publish at — see / .
+public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload, int Qos = MqttWriteEncoder.WriteQos);
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
index 306ea3c8..36f5182b 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
@@ -544,6 +544,11 @@ public sealed class MqttDriver
///
///
///
+ /// Also the write path's type (STAB-25). encodes a Sparkplug
+ /// command with this same answer, so a birth-typed Int32 metric is commanded as Int32 rather than
+ /// as the record's default.
+ ///
+ ///
/// The precedence is the ingest path's, restated. SparkplugIngestor.PublishMetric
/// coerces every value with DataTypeAuthored ? authored : birth; a discovery surface
/// that disagreed with it would advertise a type no value ever arrives as. The two must be
@@ -674,7 +679,10 @@ public sealed class MqttDriver
/// reference reject every other write in the same OPC UA request.
///
///
- /// Good means "published", not "applied". MQTT has no application-level write
+ /// Good means "published", not "applied". A Plain-mode write is published at QoS 1
+ /// (STAB-26), so "published" means the broker PUBACKed it — a broker-side refusal or a drop under
+ /// the send is Bad, not Good. Sparkplug NCMD/DCMD stay at QoS 0 (a redelivered command would be
+ /// re-applied). MQTT has no application-level write
/// acknowledgement, so a successful PUBLISH is the strongest truth available at this seam —
/// the same optimistic-Good posture Galaxy's fire-and-forget write takes. Device-level truth
/// arrives afterwards, as the echoed value on the tag's read topic, and corrects the node
@@ -709,7 +717,8 @@ public sealed class MqttDriver
continue;
}
- var error = MqttWriteEncoder.TryEncode(_options.Mode, def, w.Value, out var encoded);
+ var error = MqttWriteEncoder.TryEncode(
+ _options.Mode, def, w.Value, out var encoded, ResolveDiscoveredDataType(def));
if (error != MqttWriteEncodeError.None)
{
results[i] = new WriteResult(MapEncodeError(error));
@@ -728,7 +737,7 @@ public sealed class MqttDriver
try
{
await connection
- .PublishAsync(encoded.Topic, encoded.Payload, MqttWriteEncoder.WriteQos, retain: false, cancellationToken)
+ .PublishAsync(encoded.Topic, encoded.Payload, encoded.Qos, retain: false, cancellationToken)
.ConfigureAwait(false);
results[i] = new WriteResult(0u);
}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs
index 02c1d696..ba4a2c98 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs
@@ -66,6 +66,14 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable
/// How often the role gate is sampled. Bounds the failover republish latency.
private static readonly TimeSpan RoleSampleInterval = TimeSpan.FromSeconds(1);
+ /// Ceiling on the delay between failed claim attempts (STAB-27).
+ internal static readonly TimeSpan MaxClaimBackoff = TimeSpan.FromSeconds(30);
+
+ /// Consecutive failed claim attempts; reset by a successful claim or a stand-down. Reconcile-loop-owned.
+ private int _claimFailures;
+
+ private DateTimeOffset _nextClaimAttemptUtc;
+
private readonly MqttDriverOptions _options;
private readonly string _hostId;
private readonly string _driverId;
@@ -321,10 +329,21 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable
{
if (_roleSource.ShouldActAsPrimary)
{
+ // STAB-27: a claim that keeps failing backs off instead of redialling (and logging a
+ // Warning) every RoleSampleInterval for as long as the broker refuses it. The role is still
+ // sampled each tick — only the connect attempt waits.
+ if (_claimFailures > 0 && _timeProvider.GetUtcNow() < _nextClaimAttemptUtc)
+ {
+ return;
+ }
+
await ClaimAsync(cancellationToken).ConfigureAwait(false);
+ _claimFailures = 0;
}
else
{
+ // Standing down is not a failure, and a later re-promotion should claim at once.
+ _claimFailures = 0;
await ReleaseAsync("this node is not the redundancy Primary", cancellationToken).ConfigureAwait(false);
}
}
@@ -334,14 +353,42 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable
}
catch (Exception ex)
{
- _logger?.LogWarning(
- ex,
- "MQTT driver '{DriverId}': could not reconcile the Sparkplug primary-host claim for '{HostId}'; will retry.",
- _driverId,
- _hostId);
+ if (_roleSource.ShouldActAsPrimary)
+ {
+ _claimFailures++;
+ var delay = ClaimBackoffFor(_claimFailures);
+ _nextClaimAttemptUtc = _timeProvider.GetUtcNow() + delay;
+
+ // One Warning per ATTEMPT, and attempts are now spaced by the backoff, so this is the
+ // rate limit: 2 s, 4 s, 8 s … capped at MaxClaimBackoff rather than one line a second.
+ _logger?.LogWarning(
+ ex,
+ "MQTT driver '{DriverId}': could not claim Sparkplug primary host '{HostId}' (attempt {Attempt}); retrying in {RetryIn}.",
+ _driverId,
+ _hostId,
+ _claimFailures,
+ delay);
+ }
+ else
+ {
+ _logger?.LogWarning(
+ ex,
+ "MQTT driver '{DriverId}': could not reconcile the Sparkplug primary-host claim for '{HostId}'; will retry.",
+ _driverId,
+ _hostId);
+ }
}
}
+ /// Capped exponential delay before claim attempt N+1: 2^N seconds, at most .
+ /// Consecutive failed claim attempts so far (at least 1).
+ /// How long to wait before the next attempt.
+ internal static TimeSpan ClaimBackoffFor(int failures)
+ {
+ var seconds = Math.Pow(2, Math.Min(failures, 16));
+ return TimeSpan.FromSeconds(Math.Min(seconds, MaxClaimBackoff.TotalSeconds));
+ }
+
private async Task ClaimAsync(CancellationToken cancellationToken)
{
if (Volatile.Read(ref _connection) is not null)
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/ISqlDialect.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/ISqlDialect.cs
index d968389e..40ba4ec9 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/ISqlDialect.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/ISqlDialect.cs
@@ -107,6 +107,21 @@ public interface ISqlDialect
/// The command to configure.
void ConfigureCommand(DbCommand command);
+ ///
+ /// Gives the dialect a chance to adjust a poll-path parameter after its name and value are set and
+ /// before it is added to the command (STAB-24). Authored key / row-selector values are CLR strings,
+ /// because a tag's keyValue is configuration text — which a strictly-typed provider will not
+ /// compare against a numeric key column (Npgsql sends text, and
+ /// integer = text has no operator). A dialect that needs the server to infer the type from
+ /// context overrides this. The default does nothing, which is correct for providers that coerce
+ /// implicitly (SQL Server, ODP.NET).
+ /// Implementations must tolerate any they are handed, for the same
+ /// substituted-factory reason as .
+ ///
+ /// The parameter, with and value already set.
+ /// The authored value being bound (never ; may be null).
+ void ConfigureParameter(DbParameter parameter, object? value) { }
+
///
/// The most bound parameters this backend accepts in one command — 2098 usable of SQL Server's
/// 2100, 1000 entries in an Oracle IN expression list, 65535 on the Postgres wire protocol.
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/PostgresDialect.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/PostgresDialect.cs
index 8457d1c2..f52cf8d4 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/PostgresDialect.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/PostgresDialect.cs
@@ -64,6 +64,21 @@ public sealed class PostgresDialect : ISqlDialect
// Intentionally empty: see the summary.
}
+ ///
+ /// STAB-24: a string-valued bind is sent with so the
+ /// server infers its type from the compared column. Bound as the default text, a tag whose
+ /// keyValue is "3" against an integer key column fails with
+ /// operator does not exist: integer = text. Unknown still carries the value as data — it is
+ /// never interpolated — so the injection boundary is unchanged.
+ ///
+ /// The parameter to adjust.
+ /// The authored value.
+ public void ConfigureParameter(DbParameter parameter, object? value)
+ {
+ if (parameter is NpgsqlParameter npgsql && value is string)
+ npgsql.NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Unknown;
+ }
+
///
/// 65535 — the extended-query protocol carries the parameter count as a signed/unsigned 16-bit field
/// in Bind, so that is a protocol ceiling rather than a server setting, and Npgsql raises it
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs
index 2169fc2f..0ad565ad 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs
@@ -393,6 +393,7 @@ public sealed class SqlPollReader
// and never from ParameterNames: the marker and the bind name differ on ODP.NET.
parameter.ParameterName = plan.ParameterBindNames[i];
parameter.Value = plan.Parameters[i] ?? DBNull.Value;
+ _dialect.ConfigureParameter(parameter, plan.Parameters[i]);
command.Parameters.Add(parameter);
}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyShutdownJoinTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyShutdownJoinTests.cs
new file mode 100644
index 00000000..7b610135
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyShutdownJoinTests.cs
@@ -0,0 +1,54 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.PlcFamilies;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests;
+
+///
+/// STAB-6 / #602 — the probe loop was fire-and-forget, so shutdown cancelled and disposed while the
+/// loop could still be parked in a probe-runtime call that ignores cancellation, leaving its
+/// libplctag handle alive past teardown. Shutdown now joins the loop (bounded) first.
+///
+[Trait("Category", "Unit")]
+public sealed class AbLegacyShutdownJoinTests
+{
+ private sealed class ParkedInitTag(
+ AbLegacyTagCreateParams p, TaskCompletionSource entered, TaskCompletionSource release) : FakeAbLegacyTag(p)
+ {
+ public override async Task InitializeAsync(CancellationToken ct)
+ {
+ entered.TrySetResult();
+ await release.Task.ConfigureAwait(false); // deliberately ignores ct
+ }
+ }
+
+ /// Shutdown blocks on a probe loop parked in runtime init, then the loop disposes its probe runtime.
+ [Fact]
+ public async Task Shutdown_WaitsForProbeLoopParkedInInit_ThenProbeRuntimeIsDisposed()
+ {
+ var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var factory = new FakeAbLegacyTagFactory
+ {
+ Customise = p => new ParkedInitTag(p, entered, release),
+ };
+ var drv = new AbLegacyDriver(new AbLegacyDriverOptions
+ {
+ Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0", AbLegacyPlcFamily.Slc500)],
+ Probe = new AbLegacyProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(50), ProbeAddress = "S:0" },
+ }, "drv-join", factory);
+ await drv.InitializeAsync("{}", CancellationToken.None);
+ await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+ var shutdown = drv.ShutdownAsync(CancellationToken.None);
+
+ await Task.Delay(300);
+ shutdown.IsCompleted.ShouldBeFalse("shutdown must join the probe loop parked in runtime init");
+
+ // Positive control: releasing the parked call lets shutdown finish, with the probe runtime released.
+ release.SetResult();
+ await shutdown.WaitAsync(TimeSpan.FromSeconds(5));
+
+ factory.Tags["S:0"].Disposed.ShouldBeTrue();
+ }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasShutdownJoinTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasShutdownJoinTests.cs
new file mode 100644
index 00000000..6e07ae92
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasShutdownJoinTests.cs
@@ -0,0 +1,57 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Driver.FOCAS;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests;
+
+///
+/// STAB-23 / #602 — the probe / recycle / fixed-tree loops were fire-and-forget, so shutdown
+/// cancelled and disposed while a loop could still be parked inside a connect that ignores
+/// cancellation, and that connect then completed against a shut-down driver. Shutdown now joins
+/// the loops (bounded) before disposing the client.
+///
+[Trait("Category", "Unit")]
+public sealed class FocasShutdownJoinTests
+{
+ /// A client whose connect parks until released and deliberately ignores the cancellation token.
+ private sealed class ParkedConnectClient(TaskCompletionSource entered, TaskCompletionSource release) : FakeFocasClient
+ {
+ public override async Task ConnectAsync(FocasHostAddress address, TimeSpan timeout, CancellationToken ct)
+ {
+ entered.TrySetResult();
+ await release.Task.ConfigureAwait(false);
+ await base.ConnectAsync(address, timeout, CancellationToken.None).ConfigureAwait(false);
+ }
+ }
+
+ /// Shutdown blocks on a probe loop parked in connect, then disposes the client that connect produced.
+ [Fact]
+ public async Task Shutdown_WaitsForLoopParkedInConnect_ThenDisposesItsClient()
+ {
+ var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var factory = new FakeFocasClientFactory { Customise = () => new ParkedConnectClient(entered, release) };
+ var options = new FocasDriverOptions
+ {
+ Devices = [new FocasDeviceOptions("focas://10.0.0.5:8193")],
+ RawTags = [FocasRawTags.Entry(new FocasTagDefinition("R", "focas://10.0.0.5:8193", "R100", FocasDataType.Byte))],
+ Probe = new FocasProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(50), Timeout = TimeSpan.FromSeconds(10) },
+ };
+ var drv = new FocasDriver(options, "focas-1", factory);
+ await drv.InitializeAsync("{}", CancellationToken.None);
+ await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+ var shutdown = drv.ShutdownAsync(CancellationToken.None);
+
+ await Task.Delay(300);
+ shutdown.IsCompleted.ShouldBeFalse("shutdown must join the probe loop parked in connect");
+
+ // Positive control: releasing the parked connect lets shutdown finish.
+ release.SetResult();
+ await shutdown.WaitAsync(TimeSpan.FromSeconds(5));
+
+ factory.Clients.ShouldNotBeEmpty();
+ factory.Clients.ShouldAllBe(c => c.DisposeCount >= 1);
+ factory.Clients.ShouldAllBe(c => !c.IsConnected);
+ }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
new file mode 100644
index 00000000..738f7a5f
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
@@ -0,0 +1,98 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
+
+///
+/// STAB-20 / #602 — a Subscribe that lands between StopSampleStreamAsync clearing the pump slots
+/// and the teardown that retires the client started a pump on the soon-to-be-disposed session. The
+/// lifecycle restart then saw "a pump is running" and never started one against the new session, leaving
+/// the subscription with no live stream. The pump is now stamped with the session generation and a
+/// start request supersedes a pump from an older one.
+///
+public sealed class MTConnectPumpGenerationTests
+{
+ private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(10);
+
+ private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+ /// The pump a mid-restart Subscribe started on the retired client is replaced by one on the new client.
+ [Fact]
+ public async Task Subscribe_during_a_lifecycle_restart_rebinds_the_pump_to_the_new_client()
+ {
+ var clientA = CannedAgentClient.FromFixtures();
+ var clientB = CannedAgentClient.FromFixtures();
+ var clients = new Queue([clientA, clientB]);
+
+ var driver = new MTConnectDriver(
+ new MTConnectDriverOptions
+ {
+ AgentUri = "http://fixture-agent:5000",
+ RequestTimeoutMs = 5000,
+ Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
+ },
+ "mt-gen",
+ _ => clients.Dequeue());
+ await driver.InitializeAsync("{}", Ct);
+
+ var blockOnce = 0;
+ var blocking = new ManualResetEventSlim(false);
+ var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ driver.OnDataChange += (_, e) =>
+ {
+ // Once only: the second subscriber's initial-data callback replays the same 1.5 and must not wedge.
+ if (e.Snapshot.Value is not 1.5d || Interlocked.Exchange(ref blockOnce, 1) != 0)
+ {
+ return;
+ }
+
+ entered.TrySetResult();
+ blocking.Wait();
+ };
+
+ try
+ {
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ _ = clientA.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "1.5"));
+ await entered.Task.WaitAsync(Watchdog, Ct);
+
+ // The first pump is now wedged in caller code. A re-initialize clears the pump slots and then
+ // waits (bounded) on it, which is the window the race lives in.
+ var restart = driver.InitializeAsync("{}", Ct);
+ var spins = 0;
+ while (driver.SampleStreamTask is not null)
+ {
+ await Task.Delay(10, Ct);
+ (++spins).ShouldBeLessThan(500, "restart never cleared the pump slot");
+ }
+
+ // A second subscriber arrives mid-restart: it starts a pump on the (still-live) old client.
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ await clientA.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
+ restart.IsCompleted.ShouldBeFalse();
+
+ blocking.Set();
+ await restart.WaitAsync(Watchdog, Ct);
+
+ // The restart installed a new client; the standing subscription must be streaming from IT.
+ await clientB.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
+ clientA.IsDisposed.ShouldBeTrue();
+ }
+ finally
+ {
+ blocking.Set();
+ await driver.ShutdownAsync(CancellationToken.None);
+ }
+ }
+}
+
+internal static class MTConnectPumpGenerationChunks
+{
+ public static MTConnectStreamsResult Pos(long first, long next, string value) =>
+ new(
+ 1655000000L,
+ next,
+ first,
+ [new MTConnectObservation("dev1_pos", value, new DateTime(2026, 7, 24, 12, 30, 0, DateTimeKind.Utc))]);
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs
index 20984a5b..56784606 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs
@@ -350,6 +350,55 @@ public sealed class MqttWriteEncoderTests
Payload.Parser.ParseFrom(encoded.Payload).Metrics[0].Name.ShouldBe("Node Control/Rebirth");
}
+ // ---- STAB-25 / STAB-26 (#602) ---------------------------------------------------------------------
+
+ [Fact]
+ public void A_plain_write_publishes_at_QoS1_so_Good_means_the_broker_accepted_it()
+ {
+ MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("oven/set"), 42.5, out var encoded)
+ .ShouldBe(MqttWriteEncodeError.None);
+
+ encoded.Qos.ShouldBe(1);
+ }
+
+ [Fact]
+ public void A_sparkplug_command_stays_at_QoS0()
+ {
+ // Pinned beside the Plain case so the two cannot be "unified" by accident: an NCMD/DCMD retry from
+ // QoS 1 redelivery would re-apply a command the device already took.
+ MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, Sparkplug("Filler1"), 1.0, out var encoded)
+ .ShouldBe(MqttWriteEncodeError.None);
+
+ encoded.Qos.ShouldBe(0);
+ }
+
+ [Fact]
+ public void An_unauthored_sparkplug_type_is_commanded_with_the_birth_declared_type()
+ {
+ // DataTypeAuthored=false: the record's Float64 is only a default. The birth said Int32, so the DCMD
+ // must carry Int32 — a Double-typed command to an Int32 metric is the silent-drop shape.
+ var def = Sparkplug("Filler1") with { DataTypeAuthored = false };
+
+ MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 7, out var encoded, DriverDataType.Int32)
+ .ShouldBe(MqttWriteEncodeError.None);
+
+ var metric = Payload.Parser.ParseFrom(encoded.Payload).Metrics[0];
+ metric.Datatype.ShouldBe((uint)TahuDataType.Int32);
+ metric.IntValue.ShouldBe(7u);
+ }
+
+ [Fact]
+ public void An_authored_sparkplug_type_wins_over_the_birth_type()
+ {
+ // Same precedence as the ingest path: authored beats birth.
+ var def = Sparkplug("Filler1", DriverDataType.Float64) with { DataTypeAuthored = true };
+
+ MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 7, out var encoded, DriverDataType.Int32)
+ .ShouldBe(MqttWriteEncodeError.None);
+
+ Payload.Parser.ParseFrom(encoded.Payload).Metrics[0].Datatype.ShouldBe((uint)TahuDataType.Double);
+ }
+
private static string EncodeTopic(MqttMode mode, MqttTagDefinition def)
{
MqttWriteEncoder.TryEncode(mode, def, 1.0, out var encoded).ShouldBe(MqttWriteEncodeError.None);
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs
index 0cffd5b5..cfdcdd5a 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs
@@ -238,6 +238,53 @@ public sealed class SparkplugPrimaryHostTests
await host.DisposeAsync();
}
+ ///
+ /// STAB-27 — a claim the broker keeps refusing used to be redialled (and Warning-logged) on every
+ /// one-second role sample, forever. It now backs off. The clock is frozen, so the real one-second
+ /// loop ticks several times without the backoff window ever opening; advancing the clock opens it.
+ ///
+ [Fact]
+ public async Task A_failing_claim_backs_off_instead_of_redialling_every_tick()
+ {
+ var clock = new FrozenClock();
+ var host = new SparkplugPrimaryHost(
+ Unreachable(), "drv-1", new FixedRole(shouldActAsPrimary: true), timeProvider: clock);
+
+ await host.StartAsync(TestContext.Current.CancellationToken);
+ host.ClaimAttemptsForTests.ShouldBe(1);
+
+ // Several real loop ticks elapse while the (frozen) backoff window stays shut.
+ await Task.Delay(TimeSpan.FromMilliseconds(3200), TestContext.Current.CancellationToken);
+ host.ClaimAttemptsForTests.ShouldBe(1, "no further claim while inside the backoff window");
+
+ // Positive control: once the window passes, the very next tick does try again.
+ clock.Advance(TimeSpan.FromSeconds(5));
+ (await WaitUntilAsync(() => host.ClaimAttemptsForTests >= 2, TimeSpan.FromSeconds(10)))
+ .ShouldBeTrue("the claim must resume after the backoff window");
+
+ await host.DisposeAsync();
+ }
+
+ /// The ladder is exponential from 2 s and capped, so a long outage settles at one attempt per cap.
+ [Theory]
+ [InlineData(1, 2)]
+ [InlineData(2, 4)]
+ [InlineData(3, 8)]
+ [InlineData(4, 16)]
+ [InlineData(5, 30)]
+ [InlineData(40, 30)]
+ public void Claim_backoff_is_exponential_and_capped(int failures, int expectedSeconds)
+ => SparkplugPrimaryHost.ClaimBackoffFor(failures).ShouldBe(TimeSpan.FromSeconds(expectedSeconds));
+
+ private sealed class FrozenClock : TimeProvider
+ {
+ private DateTimeOffset _now = new(2026, 10, 8, 0, 0, 0, TimeSpan.Zero);
+
+ public override DateTimeOffset GetUtcNow() => _now;
+
+ public void Advance(TimeSpan by) => _now += by;
+ }
+
/// A broker that refuses instantly, so a connect attempt is observable but not slow.
private static MqttDriverOptions Unreachable() =>
Options() with { Host = "127.0.0.1", Port = 1, ConnectTimeoutSeconds = 1 };
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/postgres/seed-pg.sql b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/postgres/seed-pg.sql
index 21c1b520..59ce06da 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/postgres/seed-pg.sql
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/postgres/seed-pg.sql
@@ -80,6 +80,20 @@ VALUES (7, 180.5, 1.2, TIMESTAMP '2026-07-24 10:00:00'),
(9, NULL, 1.6, TIMESTAMP '2026-07-24 10:00:01'),
(8, 210.25, 1.4, TIMESTAMP '2026-07-24 10:00:02');
+-- Integer-keyed EAV source (STAB-24): the key column is `integer`, so a key bound as the default
+-- `text` fails with "operator does not exist: integer = text". Pins the untyped-bind fix in
+-- PostgresDialect.ConfigureParameter for the key-value IN-list path (the wide-row `whereColumn`
+-- path is covered by latest_status.station_id above).
+CREATE TABLE sqlpoll.int_tag_values (
+ tag_id integer NOT NULL,
+ num_value double precision NULL,
+ sample_ts timestamp(3) NOT NULL
+);
+
+INSERT INTO sqlpoll.int_tag_values (tag_id, num_value, sample_ts)
+VALUES (101, 77.5, TIMESTAMP '2026-07-24 10:00:00'),
+ (102, 12.0, TIMESTAMP '2026-07-24 10:00:01');
+
RESET ROLE;
-- ── grants: USAGE + SELECT and nothing else ───────────────────────────────────────────────────
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/PostgresReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/PostgresReadTests.cs
index fcf8f4a4..e9bc7fd3 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/PostgresReadTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/PostgresReadTests.cs
@@ -114,6 +114,63 @@ public sealed class PostgresReadTests
.ShouldBe(PostgresFixture.NewestStationOvenTemp);
}
+ // ---- integer key columns (STAB-24) ----
+
+ ///
+ /// A key-value tag whose key column is integer reads. Authored keyValue is text, and
+ /// Npgsql's default text bind has no integer = text operator — the read failed with a
+ /// provider error before PostgresDialect.ConfigureParameter sent string binds untyped.
+ ///
+ [Fact]
+ public async Task ReadAsync_realPostgres_integerKeyColumnComparesAgainstAuthoredText()
+ {
+ SkipUnlessLive();
+
+ await using var driver = await StartAsync(Tag("Sql/IntKey/Tag101", new JsonObject
+ {
+ ["driver"] = "Sql",
+ ["model"] = "KeyValue",
+ ["table"] = PostgresFixture.Qualified("int_tag_values"),
+ ["keyColumn"] = "tag_id",
+ ["keyValue"] = "101",
+ ["valueColumn"] = PostgresFixture.ValueColumn,
+ ["timestampColumn"] = PostgresFixture.TimestampColumn,
+ }));
+
+ var snapshot = (await driver.ReadAsync(["Sql/IntKey/Tag101"], TestContext.Current.CancellationToken))
+ .ShouldHaveSingleItem();
+
+ SqlStatusCodes.IsGood(snapshot.StatusCode).ShouldBeTrue();
+ Convert.ToDouble(snapshot.Value, CultureInfo.InvariantCulture).ShouldBe(77.5);
+ }
+
+ /// A wide-row whereColumn selector over the integer station_id column reads the matching row.
+ [Fact]
+ public async Task ReadAsync_realPostgres_integerWhereColumnSelectsTheMatchingStation()
+ {
+ SkipUnlessLive();
+
+ await using var driver = await StartAsync(Tag("Sql/Station7/OvenTemp", new JsonObject
+ {
+ ["driver"] = "Sql",
+ ["model"] = "WideRow",
+ ["table"] = PostgresFixture.Qualified(PostgresFixture.WideRowTable),
+ ["columnName"] = "oven_temp",
+ ["timestampColumn"] = PostgresFixture.TimestampColumn,
+ ["rowSelector"] = new JsonObject
+ {
+ ["whereColumn"] = "station_id",
+ ["whereValue"] = "7",
+ },
+ }));
+
+ var snapshot = (await driver.ReadAsync(["Sql/Station7/OvenTemp"], TestContext.Current.CancellationToken))
+ .ShouldHaveSingleItem();
+
+ SqlStatusCodes.IsGood(snapshot.StatusCode).ShouldBeTrue();
+ Convert.ToDouble(snapshot.Value, CultureInfo.InvariantCulture).ShouldBe(180.5);
+ }
+
// ---- catalog case-folding (design §8.1 gate, against a real information_schema) ----
///
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/ColonMarkerDialect.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/ColonMarkerDialect.cs
index fa8dd2b3..f514b1be 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/ColonMarkerDialect.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/ColonMarkerDialect.cs
@@ -104,6 +104,16 @@ public sealed class ColonMarkerDialect : ISqlDialect
lock (_commands) _commands.Add(command);
}
+ /// Every value handed to , in call order.
+ public List ConfiguredParameterValues { get; } = [];
+
+ ///
+ public void ConfigureParameter(DbParameter parameter, object? value)
+ {
+ ArgumentNullException.ThrowIfNull(parameter);
+ lock (ConfiguredParameterValues) ConfiguredParameterValues.Add(value);
+ }
+
///
public string QuoteIdentifier(string ident) => Inner.QuoteIdentifier(ident);
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/PostgresDialectTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/PostgresDialectTests.cs
index bfd979f8..2f56b7d1 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/PostgresDialectTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/PostgresDialectTests.cs
@@ -24,6 +24,26 @@ public class PostgresDialectTests
d.QuoteIdentifier("a\"b").ShouldBe("\"a\"\"b\"");
}
+ /// STAB-24: a string bind is untyped so Postgres infers it against an integer key column.
+ [Fact]
+ public void ConfigureParameter_sendsStringBindsAsUnknownType_soIntegerKeysCompare()
+ {
+ var p = new Npgsql.NpgsqlParameter("@k0", "3") { NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Text };
+ new PostgresDialect().ConfigureParameter(p, "3");
+ p.NpgsqlDbType.ShouldBe(NpgsqlTypes.NpgsqlDbType.Unknown);
+ }
+
+ /// Non-string values keep their natural type, and a foreign parameter type is tolerated.
+ [Fact]
+ public void ConfigureParameter_leavesNonStringsAndForeignParametersAlone()
+ {
+ var d = new PostgresDialect();
+ var n = new Npgsql.NpgsqlParameter("@k0", 3) { NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Integer };
+ d.ConfigureParameter(n, 3);
+ n.NpgsqlDbType.ShouldBe(NpgsqlTypes.NpgsqlDbType.Integer);
+ Should.NotThrow(() => d.ConfigureParameter(new SqliteParameter("@k0", "3"), "3"));
+ }
+
[Fact]
public void QuoteIdentifier_rejectsControlCharsAndNul()
=> Should.Throw(() => new PostgresDialect().QuoteIdentifier("x\0y"));
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlCatalogGateDriverTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlCatalogGateDriverTests.cs
index f94714c3..f0e5d373 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlCatalogGateDriverTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlCatalogGateDriverTests.cs
@@ -318,6 +318,35 @@ public sealed class SqlCatalogGateDriverTests
dialect.ConfiguredCommands.ShouldBeGreaterThan(0);
}
+ ///
+ /// STAB-24: every poll-path bind goes through with the
+ /// authored value, so a dialect can type it. Without the call the Postgres untyped-bind fix is inert.
+ ///
+ [Fact]
+ public async Task The_poll_binds_every_authored_value_through_the_dialects_parameter_hook()
+ {
+ using var fixture = new SqlitePollFixture();
+ var dialect = new ColonMarkerDialect();
+ await using var driver = new SqlDriver(
+ new SqlDriverOptions
+ {
+ RawTags = [KvEntry("Speed", SqlitePollFixture.PresentKey)],
+ OperationTimeout = TimeSpan.FromSeconds(15),
+ CommandTimeout = TimeSpan.FromSeconds(10),
+ },
+ DriverInstanceId,
+ dialect,
+ fixture.ConnectionString,
+ factory: fixture.Factory,
+ logger: NullLogger.Instance);
+ await driver.InitializeAsync(ConfigJson, CancellationToken.None);
+
+ var snapshots = await driver.ReadAsync(["Speed"], CancellationToken.None);
+
+ SqlStatusCodes.IsGood(snapshots[0].StatusCode).ShouldBeTrue();
+ dialect.ConfiguredParameterValues.ShouldContain(SqlitePollFixture.PresentKey);
+ }
+
// ---- helpers ----
///