diff --git a/docker-dev/seed/seed-clusters.sql b/docker-dev/seed/seed-clusters.sql index 62bfa596..97ceca09 100644 --- a/docker-dev/seed/seed-clusters.sql +++ b/docker-dev/seed/seed-clusters.sql @@ -106,6 +106,28 @@ IF NOT EXISTS (SELECT 1 FROM dbo.ClusterNode WHERE NodeId = 'site-b-2:4053') (NodeId, ClusterId, Host, OpcUaPort, DashboardPort, AkkaPort, GrpcPort, ApplicationUri, ServiceLevelBase, Enabled, CreatedBy) VALUES ('site-b-2:4053', 'SITE-B', 'site-b-2', 4840, 8081, 4053, 4056, 'urn:OtOpcUa:site-b-2', 150, 1, 'docker-dev-seed'); +------------------------------------------------------------------------------ +-- Backfill: GrpcPort on rows that predate the column (#493) +-- +-- Every insert above is INSERT-if-not-exists, which is the right shape for a re-runnable seed +-- but means a *new nullable* column never reaches rows created by an earlier version of this +-- file. On a persisted volume that is exactly what happened when Phase 5 added GrpcPort: the +-- six ClusterNode rows already existed, so they kept NULL, TelemetryNodeSource found no dial +-- targets, and TelemetryDial:Mode=Grpc silently connected to nothing (throttled Warning per +-- node — the code degrades gracefully, so nothing failed loudly). +-- +-- Backfills NULL only. It must not overwrite a port an operator set deliberately on a +-- long-lived dev volume; a NULL is the unambiguous "this row predates the column" marker. +-- +-- AkkaPort deliberately has no equivalent here: it is NOT NULL with a default of 4053, so +-- pre-existing rows already carry a usable value and there is nothing to backfill. +------------------------------------------------------------------------------ + +UPDATE dbo.ClusterNode + SET GrpcPort = 4056 -- matches Telemetry__GrpcListenPort on all six nodes in docker-compose.yml + WHERE GrpcPort IS NULL + AND CreatedBy = 'docker-dev-seed'; + ------------------------------------------------------------------------------ -- Galaxy MxAccess gateway — INTENTIONALLY NOT SEEDED (retired 2026-06-15) -- diff --git a/docs/Telemetry.md b/docs/Telemetry.md index a24570f1..9f04a30d 100644 --- a/docs/Telemetry.md +++ b/docs/Telemetry.md @@ -128,6 +128,39 @@ transports. ScadaBridge itself has since closed that gap with this identical interceptor pattern, so Phase 5 ships authenticated from the start rather than deferring auth to a later hardening pass. See the design doc's superseded note for the full history. +## Upgrading an existing deployment: populate `ClusterNode.GrpcPort` first (#493) + +`GrpcPort` is a **nullable column added by Phase 5**, and nothing backfills it onto rows that +already existed. On an upgraded deployment every `ClusterNode` row therefore carries `NULL`, +central finds **no dial targets**, and flipping `TelemetryDial:Mode` to `Grpc` connects to +nothing. Fresh installs are unaffected. `AkkaPort` did not have this problem only because it is +`NOT NULL` with a default of 4053. + +Nothing fails loudly, which is what makes this worth stating up front — the code degrades +gracefully, once per node, throttled: + +``` +ClusterNode has no GrpcPort; it exposes no telemetry stream and is skipped in this dial refresh +(further skips of this node are silent) +``` + +Other symptoms: no `ESTABLISHED` sockets on the telemetry port (a `LISTEN` but nothing else in +`docker exec cat /proc/net/tcp6`), and the `/alerts` / `/script-log` pill never turning +live in `Grpc` mode. + +**Before flipping any node to `Grpc`**, set each driver node's `GrpcPort` to that node's own +`Telemetry:GrpcListenPort` — via the AdminUI node editor, or directly: + +```sql +UPDATE dbo.ClusterNode SET GrpcPort = WHERE NodeId = ''; +``` + +There is deliberately **no EF data migration** backfilling this. The correct value is that node's +own listener port, which is per-deployment configuration the database cannot derive; a migration +could only invent one, and a *wrong* dial target is worse than the `NULL` the dialer already skips +explicitly. The docker-dev seed does backfill (`GrpcPort IS NULL AND CreatedBy = 'docker-dev-seed'` +→ 4056) because there the port is fixed by `docker-compose.yml` and therefore actually knowable. + ## Reconnect and reliability Central's `TelemetryDialSupervisor` (an admin-role actor) runs **one reconnecting dialer per diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs index 5b9bc7c7..fde97dbd 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs @@ -60,6 +60,11 @@ public static class DraftValidator /// credential in the ConfigDb — persisted, replicated to every node's artifact cache, and readable by /// anyone with config access — while the runtime silently ignored it and the driver failed to connect. /// Discarding a secret on read is not the same as refusing to store it. + /// Two of the three surfaces are checked here. The third — a node's + /// — is not reachable from + /// and is gated at its save instead (#499); see + /// for why a deploy gate is the wrong instrument for a value that + /// never enters the artifact. /// Checked on both config surfaces a Sql driver reads: the instance's /// and the of every device /// beneath it, because the two are merged before the DTO sees them — a credential pasted into the @@ -74,7 +79,7 @@ public static class DraftValidator /// private static void ValidateSqlConnectionStringNotPersisted(DraftSnapshot draft, List errors) { - const string ForbiddenKey = "connectionString"; + const string ForbiddenKey = SqlCredentialGuard.ForbiddenKey; var sqlInstanceIds = draft.DriverInstances .Where(d => string.Equals(d.DriverType, Core.Abstractions.DriverTypeNames.Sql, StringComparison.Ordinal)) @@ -85,7 +90,7 @@ public static class DraftValidator foreach (var d in draft.DriverInstances) { if (!sqlInstanceIds.Contains(d.DriverInstanceId)) continue; - if (!HasTopLevelKey(d.DriverConfig, ForbiddenKey)) continue; + if (!SqlCredentialGuard.CarriesLiteralConnectionString(d.DriverConfig)) continue; errors.Add(new("SqlConnectionStringPersisted", $"Sql driver instance '{d.DriverInstanceId}' has a '{ForbiddenKey}' key in its DriverConfig. " + "A Sql driver must name its credentials indirectly via 'connectionStringRef', which resolves " + @@ -98,7 +103,7 @@ public static class DraftValidator foreach (var dev in draft.Devices) { if (!sqlInstanceIds.Contains(dev.DriverInstanceId)) continue; - if (!HasTopLevelKey(dev.DeviceConfig, ForbiddenKey)) continue; + if (!SqlCredentialGuard.CarriesLiteralConnectionString(dev.DeviceConfig)) continue; errors.Add(new("SqlConnectionStringPersisted", $"Device '{dev.DeviceId}' on Sql driver instance '{dev.DriverInstanceId}' has a " + $"'{ForbiddenKey}' key in its DeviceConfig. DeviceConfig is merged onto DriverConfig before " + @@ -107,34 +112,6 @@ public static class DraftValidator } } - /// - /// True when is a JSON object carrying at its top level, - /// matched case-insensitively. Never throws — a blank, malformed or non-object blob simply has no keys, - /// and shaping the config JSON is another rule's job. - /// - /// The config blob to inspect. - /// The property name to look for. - /// when the key is present at the top level. - private static bool HasTopLevelKey(string? json, string key) - { - if (string.IsNullOrWhiteSpace(json)) return false; - try - { - using var doc = System.Text.Json.JsonDocument.Parse(json); - if (doc.RootElement.ValueKind != System.Text.Json.JsonValueKind.Object) return false; - foreach (var property in doc.RootElement.EnumerateObject()) - { - if (string.Equals(property.Name, key, StringComparison.OrdinalIgnoreCase)) return true; - } - - return false; - } - catch (System.Text.Json.JsonException) - { - return false; - } - } - /// WP7 Calculation-driver deploy gates. For every tag bound to a Calculation driver: /// /// scriptId existence — the tag's TagConfig.scriptId must be present and resolve to diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/SqlCredentialGuard.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/SqlCredentialGuard.cs new file mode 100644 index 00000000..11ed396d --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/SqlCredentialGuard.cs @@ -0,0 +1,139 @@ +using System.Text.Json; + +namespace ZB.MOM.WW.OtOpcUa.Configuration.Validation; + +/// +/// The one place that decides whether a persisted config blob carries a literal Sql +/// connectionString (Gitea #498 / #499). +/// +/// +/// +/// A Sql driver names its credentials indirectly — connectionStringRef resolves from +/// the environment / secret store at Initialize — so nothing persisted should ever contain a +/// database password. The typed SqlDriverConfigDto already drops a connectionString +/// key on read, but discarding a secret on read is not the same as refusing to store it: +/// config blobs are schemaless JSON columns authored through raw-JSON textareas, so the key can +/// be written even though the runtime ignores it. +/// +/// +/// There are three surfaces a Sql driver's config is persisted on, and they need different +/// enforcement points: +/// +/// DriverInstance.DriverConfig and Device.DeviceConfig — both in the +/// deployed artifact, both visible to DraftSnapshot, both gated by +/// DraftValidator. +/// ClusterNode.DriverConfigOverridesJson — the per-node override map. It is +/// not in DraftSnapshot, and deliberately does not go there: adding a +/// ClusterNode collection would widen the snapshot (and every builder of one) for a +/// single rule about a value that is not part of the artifact at all. More to the point, a +/// deploy gate is the wrong instrument here — the artifact never carries node overrides, so +/// blocking a deploy would not stop the credential being stored. It is already in the +/// database by then. This surface is therefore gated at the save, which is the only +/// point where "refuse to store it" is literally true. +/// +/// +/// +/// Deliberately narrow. This checks the Sql driver's connectionString key +/// only — not a general "credential-shaped key" sweep over every driver type. A broader rule +/// would start refusing configs that are legitimate today for drivers that never made this +/// guarantee, turning defence-in-depth into a regression. Widen it per driver, as each driver +/// gains an indirect-credential contract of its own. +/// +/// +public static class SqlCredentialGuard +{ + /// The config key a Sql driver must never persist. Matched case-insensitively. + public const string ForbiddenKey = "connectionString"; + + /// + /// Whether is a JSON object carrying at + /// its top level. + /// + /// + /// Matched case-insensitively: System.Text.Json binds ConnectionString to a + /// connectionString property by default, so a case variant is the same key, not a + /// different one. Only the top level is scanned — the DTO is flat, so a nested occurrence cannot + /// bind and is not the credential-shaped mistake this exists to catch. + /// + /// The config blob to inspect. + /// when the key is present at the top level. + public static bool CarriesLiteralConnectionString(string? configJson) + { + if (string.IsNullOrWhiteSpace(configJson)) return false; + try + { + using var doc = JsonDocument.Parse(configJson); + return doc.RootElement.ValueKind == JsonValueKind.Object + && HasTopLevelKey(doc.RootElement, ForbiddenKey); + } + catch (JsonException) + { + // A malformed blob simply has no keys. Shaping the config JSON is another rule's job, and + // throwing here would turn a formatting mistake into a credential-guard failure. + return false; + } + } + + /// + /// The DriverInstanceIds in a ClusterNode.DriverConfigOverridesJson map whose + /// override object carries a literal connection string, restricted to instances that are actually + /// Sql drivers. + /// + /// + /// The override map is shaped { "<DriverInstanceId>": { …driver config keys… } } and is + /// merged onto the cluster-level DriverConfig, so a credential pasted into one lands in + /// exactly the place already refuses on the driver + /// itself. + /// + /// The node's override map. Null / blank / malformed yields no + /// violations. + /// The ids of driver instances whose DriverType is + /// Sql. Keys outside this set are ignored — a non-Sql driver never made the indirect-credential + /// guarantee, so flagging it would be a regression, not defence in depth. + /// The offending driver-instance ids, in document order; empty when there are none. + public static IReadOnlyList FindNodeOverrideViolations( + string? overridesJson, IReadOnlyCollection sqlDriverInstanceIds) + { + if (string.IsNullOrWhiteSpace(overridesJson) || sqlDriverInstanceIds.Count == 0) + { + return []; + } + + var sqlIds = sqlDriverInstanceIds as IReadOnlySet + ?? sqlDriverInstanceIds.ToHashSet(StringComparer.Ordinal); + + try + { + using var doc = JsonDocument.Parse(overridesJson); + if (doc.RootElement.ValueKind != JsonValueKind.Object) return []; + + List? offenders = null; + foreach (var entry in doc.RootElement.EnumerateObject()) + { + if (!sqlIds.Contains(entry.Name)) continue; + if (entry.Value.ValueKind != JsonValueKind.Object) continue; + if (!HasTopLevelKey(entry.Value, ForbiddenKey)) continue; + + (offenders ??= []).Add(entry.Name); + } + + return (IReadOnlyList?)offenders ?? []; + } + catch (JsonException) + { + return []; + } + } + + /// Whether carries at its top level, + /// case-insensitively. + private static bool HasTopLevelKey(JsonElement obj, string key) + { + foreach (var property in obj.EnumerateObject()) + { + if (string.Equals(property.Name, key, StringComparison.OrdinalIgnoreCase)) return true; + } + + return false; + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Clusters/NodeEdit.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Clusters/NodeEdit.razor index 2c1fef76..a246c14b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Clusters/NodeEdit.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Clusters/NodeEdit.razor @@ -9,6 +9,8 @@ @using System.ComponentModel.DataAnnotations @using ZB.MOM.WW.OtOpcUa.Configuration @using ZB.MOM.WW.OtOpcUa.Configuration.Entities +@using ZB.MOM.WW.OtOpcUa.Configuration.Validation +@using ZB.MOM.WW.OtOpcUa.Core.Abstractions @inject IDbContextFactory DbFactory @inject NavigationManager Nav @inject AuthenticationStateProvider AuthState @@ -34,6 +36,10 @@ else { + @* Without this, a validation failure suppresses OnValidSubmit with NO feedback whatsoever — + the Save button simply does nothing and the operator has no way to tell why. That is how the + NodeId-regex defect above stayed invisible. *@ +
Identity
@@ -178,6 +184,33 @@ else } await using var db = await DbFactory.CreateDbContextAsync(); + + // #499 — the third Sql-credential surface. DriverConfigOverridesJson is merged onto the + // cluster-level DriverConfig, so a literal connectionString pasted here leaks exactly as one + // pasted into the driver itself. The deploy gate cannot see it (DraftSnapshot carries no + // ClusterNode rows) and would be the wrong place anyway: node overrides never enter the + // artifact, so by the time a deploy ran the credential would already be stored. Refuse the + // save instead. The message names the driver instance but NEVER the value. + if (!string.IsNullOrWhiteSpace(_form.DriverConfigOverridesJson)) + { + var sqlDriverIds = await db.DriverInstances + .Where(d => d.ClusterId == ClusterId && d.DriverType == DriverTypeNames.Sql) + .Select(d => d.DriverInstanceId) + .ToListAsync(); + + var offenders = SqlCredentialGuard.FindNodeOverrideViolations( + _form.DriverConfigOverridesJson, sqlDriverIds); + if (offenders.Count > 0) + { + _error = + $"Driver config overrides carry a '{SqlCredentialGuard.ForbiddenKey}' for Sql driver " + + $"instance(s) {string.Join(", ", offenders)}. A Sql driver must name its credentials " + + "indirectly via 'connectionStringRef', which resolves from the environment / secret " + + "store at Initialize; a literal connection string here would be stored in the config " + + "database, and the runtime ignores it anyway. Remove the key."; + return; + } + } if (IsNew) { if (await db.ClusterNodes.AnyAsync(n => n.NodeId == _form.NodeId)) @@ -194,7 +227,7 @@ else OpcUaPort = _form.OpcUaPort, DashboardPort = _form.DashboardPort, ApplicationUri = _form.ApplicationUri, - ServiceLevelBase = _form.ServiceLevelBase, + ServiceLevelBase = (byte)_form.ServiceLevelBase, Enabled = _form.Enabled, DriverConfigOverridesJson = string.IsNullOrWhiteSpace(_form.DriverConfigOverridesJson) ? null : _form.DriverConfigOverridesJson, CreatedAt = DateTime.UtcNow, @@ -210,7 +243,7 @@ else entity.OpcUaPort = _form.OpcUaPort; entity.DashboardPort = _form.DashboardPort; entity.ApplicationUri = _form.ApplicationUri; - entity.ServiceLevelBase = _form.ServiceLevelBase; + entity.ServiceLevelBase = (byte)_form.ServiceLevelBase; entity.Enabled = _form.Enabled; entity.DriverConfigOverridesJson = string.IsNullOrWhiteSpace(_form.DriverConfigOverridesJson) ? null : _form.DriverConfigOverridesJson; } @@ -254,14 +287,26 @@ else private sealed class FormModel { - [Required, RegularExpression("^[A-Za-z0-9_-]+$")] + // The ':' and '.' are REQUIRED, not permissive. Every NodeId in this system is "host:port" — + // ClusterRoleInfo and ConfigPublishCoordinator derive it as member.Address.Host:Port, the seed + // writes "central-1:4053", and NodeDeploymentState.NodeId is FK-bound to it. The previous + // pattern forbade the colon, so EVERY existing node failed validation and OnValidSubmit never + // ran: the Save button did nothing at all, silently. Hostnames may also be dotted + // (line3-opc-a.plant.local:4053). + [Required, RegularExpression("^[A-Za-z0-9_.:-]+$")] public string NodeId { get; set; } = ""; [Required] public string Host { get; set; } = ""; [Range(1, 65535)] public int OpcUaPort { get; set; } = 4840; [Range(1, 65535)] public int DashboardPort { get; set; } = 8081; [Required, RegularExpression("^urn:[A-Za-z0-9_:./-]+$")] public string ApplicationUri { get; set; } = ""; - [Range(0, 255)] public byte ServiceLevelBase { get; set; } = 200; + // int, NOT byte. Blazor's InputNumber rejects System.Byte outright — its static + // constructor throws "The type 'System.Byte' is not a supported numeric type", which escapes + // BuildRenderTree unhandled and takes the WHOLE page to an HTTP 500. Binding this field to + // the entity's own byte type meant /clusters/{id}/nodes/{nodeId} and .../nodes/new never + // rendered at all. The Range attribute still holds it to a byte's domain, and SubmitAsync + // casts on the way to the entity, so the stored type is unchanged. + [Range(0, 255)] public int ServiceLevelBase { get; set; } = 200; public bool Enabled { get; set; } = true; public string? DriverConfigOverridesJson { get; set; } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 3373c182..c0ca7f97 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -170,6 +170,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Interval between bounded post-connect re-discovery passes. Production default 2s; tests /// inject a tiny value so the loop runs without real-time waits. private readonly TimeSpan _rediscoverInterval; + private readonly TimeSpan _healthPollInterval; /// Cap on the number of post-connect re-discovery passes — a backstop so a never-stabilising /// (or perpetually-empty) discovered set cannot spin the loop forever. Production default 15. @@ -239,6 +240,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Optional per-pass timeout for ; defaults to 30 seconds. /// Optional Phase 6.1 resilience invoker wrapping this driver's capability /// calls; defaults to (pass-through) when not supplied. + /// Optional period of the health-poll heartbeat, which also drives the + /// subscription reconcile (); defaults to + /// . Exists so a test can drive the reconcile without waiting 30 s. /// A instance configured to create the wrapped . public static Props Props( IDriver driver, @@ -249,7 +253,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers TimeSpan? rediscoverInterval = null, int rediscoverMaxAttempts = DefaultRediscoverMaxAttempts, TimeSpan? rediscoverDiscoverTimeout = null, - IDriverCapabilityInvoker? invoker = null) => + IDriverCapabilityInvoker? invoker = null, + TimeSpan? healthPollInterval = null) => Akka.Actor.Props.Create(() => new DriverInstanceActor( driver, reconnectInterval ?? DefaultReconnectInterval, @@ -259,7 +264,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers rediscoverInterval, rediscoverMaxAttempts, rediscoverDiscoverTimeout, - invoker)); + invoker, + healthPollInterval)); /// /// Returns true when the driver should boot in DEV-STUB mode based on host platform and @@ -293,6 +299,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Per-pass timeout for ; defaults to 30 seconds. /// Phase 6.1 resilience invoker wrapping this driver's capability calls; /// defaults to (pass-through) when null. + /// Period of the health-poll heartbeat, which also drives the + /// subscription reconcile; defaults to when null. public DriverInstanceActor( IDriver driver, TimeSpan reconnectInterval, @@ -302,7 +310,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers TimeSpan? rediscoverInterval = null, int rediscoverMaxAttempts = DefaultRediscoverMaxAttempts, TimeSpan? rediscoverDiscoverTimeout = null, - IDriverCapabilityInvoker? invoker = null) + IDriverCapabilityInvoker? invoker = null, + TimeSpan? healthPollInterval = null) { _driver = driver; _invoker = invoker ?? NullDriverCapabilityInvoker.Instance; @@ -313,6 +322,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers _rediscoverInterval = rediscoverInterval ?? DefaultRediscoverInterval; _rediscoverMaxAttempts = rediscoverMaxAttempts; _rediscoverDiscoverTimeout = rediscoverDiscoverTimeout ?? DefaultRediscoverDiscoverTimeout; + _healthPollInterval = healthPollInterval ?? HealthPollInterval; OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, new KeyValuePair("event", startStubbed ? "spawn_stub" : "spawn"), new KeyValuePair("driver_type", driver.DriverType)); @@ -335,7 +345,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // actor starts, before any state transition fires. Also start the periodic heartbeat so // long-lived Healthy drivers keep their snapshot fresh for newly-joined SignalR clients. PublishHealthSnapshot(); - Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, HealthPollInterval); + Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, _healthPollInterval); } private void Stubbed() @@ -489,7 +499,58 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // to Self (HandleSubscribeAsync already logged the underlying cause). Swallow it so it doesn't dead-letter. Receive(msg => _log.Debug("DriverInstance {Id}: resubscribe reported failure: {Reason}", _driverInstanceId, msg.Reason)); - Receive(_ => PublishHealthSnapshot()); + Receive(_ => + { + PublishHealthSnapshot(); + ReconcileSubscription(); + }); + } + + /// + /// Periodic desired-vs-actual subscription reconcile (#488), riding the existing health-poll + /// tick. While Connected, a driver that has a desired subscription set but no live handle is + /// re-subscribed. + /// + /// + /// + /// The desired set is otherwise only (re)applied on a Connected transition + /// () or on a deploy's . + /// Nothing re-asserted it for a subscription that was lost while the driver stayed + /// Connected — a failed subscribe whose retry never came, or a handle dropped down an error + /// path. That is the hole this closes. + /// + /// + /// This is a state-machine consistency check, not a probe. It can only see that this + /// actor holds no handle; it cannot detect a handle that is present but dead server-side, + /// because ISubscriptionHandle is opaque (a DiagnosticId string and nothing + /// else). Detecting that needs a liveness member on ISubscribable implemented across + /// all eight drivers, each with different subscription mechanics — deliberately deferred + /// until a real occurrence shows the handle outliving the subscription. + /// + /// + /// Gated on : a driver that cannot subscribe at all may still have + /// been handed a desired set, and re-telling Subscribe to it every tick would fail + /// every tick, forever. + /// + /// + /// A redundant Subscribe is possible but harmless: a tick sitting in the mailbox ahead + /// of an already-queued Subscribe observes the same "no handle" state and tells a + /// second one. is a ReceiveAsync (so no tick can be + /// processed mid-subscribe) and drops any prior subscription before establishing the new one, + /// so the duplicate costs one extra round-trip and converges. Suppressing it would need a + /// pending-subscribe flag threaded through every self-tell site — more state than the race + /// is worth. + /// + /// + private void ReconcileSubscription() + { + if (_driver is not ISubscribable) return; + if (_desiredRefs.Count == 0 || _subscriptionHandle is not null) return; + + _log.Info( + "DriverInstance {Id}: connected with {Count} desired subscription ref(s) but no live subscription handle — re-subscribing (periodic reconcile)", + _driverInstanceId, _desiredRefs.Count); + Self.Tell(new Subscribe(_desiredRefs, _desiredInterval)); } private void Reconnecting() diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/SqlCredentialGuardTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/SqlCredentialGuardTests.cs new file mode 100644 index 00000000..6cc1e749 --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/SqlCredentialGuardTests.cs @@ -0,0 +1,132 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Configuration.Validation; + +namespace ZB.MOM.WW.OtOpcUa.Configuration.Tests; + +/// +/// — the shared "a Sql driver never persists a literal connection +/// string" check behind both the #498 deploy gate and the #499 node-override save gate. +/// +/// +/// The node-override half is the one #499 is about: ClusterNode.DriverConfigOverridesJson is a +/// map keyed by DriverInstanceId that is merged onto the cluster-level DriverConfig, and +/// it is invisible to DraftSnapshot — so nothing gated it at all before. +/// +[Trait("Category", "Unit")] +public sealed class SqlCredentialGuardTests +{ + /// The literal an operator would paste into a raw-JSON textarea. + private const string Leaked = + """{"provider":"SqlServer","connectionString":"Server=sql,1433;Database=Mes;User ID=sa;Password=hunter2"}"""; + + private const string Clean = """{"provider":"SqlServer","connectionStringRef":"Sql:Line3"}"""; + + // ---- CarriesLiteralConnectionString -------------------------------------------------------- + + /// The key is caught at the top level of a config blob. + [Fact] + public void A_literal_connectionString_is_detected() + => SqlCredentialGuard.CarriesLiteralConnectionString(Leaked).ShouldBeTrue(); + + /// The supported indirect form is not a violation — otherwise the rule would block the fix + /// it tells operators to apply. + [Fact] + public void The_indirect_connectionStringRef_form_is_clean() + => SqlCredentialGuard.CarriesLiteralConnectionString(Clean).ShouldBeFalse(); + + /// System.Text.Json binds ConnectionString to a connectionString property by + /// default, so a case variant is the same key — not a bypass. + [Theory] + [InlineData("""{"ConnectionString":"Server=x;Password=p"}""")] + [InlineData("""{"CONNECTIONSTRING":"Server=x;Password=p"}""")] + [InlineData("""{"connectionstring":"Server=x;Password=p"}""")] + public void Case_variants_are_the_same_key(string json) + => SqlCredentialGuard.CarriesLiteralConnectionString(json).ShouldBeTrue(); + + /// Blank, malformed and non-object blobs simply have no keys. Shaping the config JSON is + /// another rule's job, and throwing here would turn a formatting mistake into a credential failure. + /// + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData(" ")] + [InlineData("{ not json")] + [InlineData("[1,2,3]")] + [InlineData("\"a string\"")] + public void Blank_malformed_and_non_object_blobs_are_not_violations(string? json) + => SqlCredentialGuard.CarriesLiteralConnectionString(json).ShouldBeFalse(); + + /// Only the top level is scanned. The DTO is flat, so a nested occurrence cannot bind and is + /// not the credential-shaped mistake this rule exists to catch. + [Fact] + public void A_nested_occurrence_is_not_flagged() + => SqlCredentialGuard + .CarriesLiteralConnectionString("""{"nested":{"connectionString":"Server=x"}}""") + .ShouldBeFalse(); + + // ---- FindNodeOverrideViolations (#499) ----------------------------------------------------- + + /// The #499 leak: a credential pasted into a node's per-driver override map. + [Fact] + public void A_credential_in_a_node_override_for_a_Sql_driver_is_a_violation() + { + var overrides = $$"""{"di-sql": {{Leaked}} }"""; + + SqlCredentialGuard.FindNodeOverrideViolations(overrides, ["di-sql"]) + .ShouldBe(["di-sql"]); + } + + /// Every offending instance is reported, not just the first — an operator fixing one at a + /// time would otherwise need as many save attempts as there are leaks. + [Fact] + public void Every_offending_instance_is_reported() + { + var overrides = $$"""{"di-a": {{Leaked}}, "di-clean": {{Clean}}, "di-b": {{Leaked}} }"""; + + SqlCredentialGuard.FindNodeOverrideViolations(overrides, ["di-a", "di-b", "di-clean"]) + .ShouldBe(["di-a", "di-b"]); + } + + /// Keys outside the Sql set are ignored. A non-Sql driver never made the + /// indirect-credential guarantee, so flagging it would break a config that is legitimate today — + /// a regression, not defence in depth. + [Fact] + public void An_override_for_a_non_Sql_driver_is_ignored() + { + var overrides = $$"""{"di-modbus": {{Leaked}} }"""; + + SqlCredentialGuard.FindNodeOverrideViolations(overrides, ["di-sql"]).ShouldBeEmpty(); + } + + /// The realistic override — a node-specific endpoint — must keep saving. + [Fact] + public void A_normal_override_is_clean() + { + var overrides = """{"di-sql": {"commandTimeoutSeconds": 45}}"""; + + SqlCredentialGuard.FindNodeOverrideViolations(overrides, ["di-sql"]).ShouldBeEmpty(); + } + + /// Nothing to check when the cluster has no Sql drivers, or the node has no overrides. + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData(" ")] + [InlineData("{ not json")] + [InlineData("[1,2,3]")] + public void Blank_and_malformed_override_maps_yield_no_violations(string? overrides) + => SqlCredentialGuard.FindNodeOverrideViolations(overrides, ["di-sql"]).ShouldBeEmpty(); + + /// An empty Sql-driver set short-circuits — there is no instance the key could belong to. + [Fact] + public void No_Sql_drivers_means_no_violations() + => SqlCredentialGuard.FindNodeOverrideViolations($$"""{"di-sql": {{Leaked}} }""", []) + .ShouldBeEmpty(); + + /// A non-object override entry cannot carry config keys, and must not throw. + [Fact] + public void A_non_object_override_entry_is_skipped() + => SqlCredentialGuard.FindNodeOverrideViolations("""{"di-sql": "connectionString"}""", ["di-sql"]) + .ShouldBeEmpty(); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpBoundedChannelTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpBoundedChannelTests.cs index 9790bbcd..6ae7f2ff 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpBoundedChannelTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpBoundedChannelTests.cs @@ -10,59 +10,109 @@ using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; /// -/// PR 6.2 — pins the EventPump's bounded-channel + drop-newest behavior. We -/// hold the dispatch loop with a slow handler so the channel fills, then verify -/// the producer keeps reading from the gw stream and increments the -/// galaxy.events.dropped counter rather than blocking. +/// PR 6.2 — pins the EventPump's bounded-channel + drop-newest behavior. We hold the dispatch loop +/// with a blocking handler so the channel fills, then verify the producer keeps reading from the gw +/// stream and increments the galaxy.events.dropped counter rather than blocking. /// +/// +/// These tests observe two background loops (RunAsync / DispatchLoopAsync) through a +/// process-wide , so every wait here is a presence assertion polled up to +/// — never a fixed sleep. A fixed sleep is a bet that both loops get scheduled +/// within it, and #503 is what losing that bet looks like when the box is busy running other test +/// assemblies. +/// public sealed class EventPumpBoundedChannelTests { - /// Verifies that the event pump drops newest events when the bounded channel fills and records metrics for dropped events. + /// Upper bound on how long a poll waits for a background loop to reach its terminal state. + /// Generous on purpose: it is spent only when something is genuinely broken, and it cannot turn a + /// failing assertion into a passing one. + private static readonly TimeSpan Budget = TimeSpan.FromSeconds(15); + + /// + /// With the dispatch loop genuinely held and the channel therefore full, the producer keeps + /// draining the gw stream and counts every rejected write on galaxy.events.dropped — + /// newest-dropped, not back-pressure. + /// + /// + /// + /// The arrangement is staged so the arithmetic is exact rather than scheduler-dependent + /// (#503): emit ONE event, wait until the dispatcher has actually entered the handler (so it + /// holds that event and the channel is empty), and only then emit the remaining nine. The + /// channel accepts capacity of them and the rest must be dropped — + /// 10 − 1 held − 2 buffered = 7, every run. + /// + /// [Fact] public async Task Drops_newest_when_channel_fills_and_records_metric() { - var counters = StartMeterCapture(); + const int totalEvents = 10; + const int channelCapacity = 2; + // One event is held inside the handler; `channelCapacity` more fit in the channel; the rest are dropped. + const int expectedDropped = totalEvents - 1 - channelCapacity; + + // Unique per run: the counters are filtered to this exact galaxy.client tag, which is what keeps a + // parallel sibling test's pump out of them (#503). A literal name would work today and break the + // day someone reuses it. + var clientName = $"PumpTest-{Guid.NewGuid():N}"; + var counters = StartMeterCapture(clientName); + + // A SYNCHRONOUS gate, deliberately. OnDataChange is EventHandler — void-returning — so an + // `async (_, _) => await gate` lambda is async void: it returns to the dispatch loop at the first + // await and holds nothing. That is exactly what #503 was: the loop drained freely, drops happened + // only when the producer happened to outrun the consumer, and the assertions rode on scheduling + // luck. Blocking the loop's own thread is what makes "the dispatcher is held" true. + var dispatchGate = new ManualResetEventSlim(false); + var handlerEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + EventPump? pump = null; try { var subscriber = new ManualSubscriber(); var registry = new SubscriptionRegistry(); registry.Register(1, [new TagBinding("Tag.A", ItemHandle: 7)]); - // Tiny channel + slow handler ⇒ producer hits FullMode.Wait → TryWrite false - // for every overflow event. - var dispatchGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - await using var pump = new EventPump( - subscriber, registry, channelCapacity: 2, clientName: "PumpTest"); - pump.OnDataChange += async (_, _) => + pump = new EventPump( + subscriber, registry, channelCapacity: channelCapacity, clientName: clientName); + pump.OnDataChange += (_, _) => { - // Block the dispatch loop until we've shoved enough events through to - // overflow the bounded channel. Consume the gate exactly once. - await dispatchGate.Task.ConfigureAwait(false); + handlerEntered.TrySetResult(); + dispatchGate.Wait(); }; pump.Start(); - const int totalEvents = 10; - for (var i = 0; i < totalEvents; i++) + // Stage 1 — one event, then wait for the dispatcher to take it and block. After this the + // channel is provably empty and exactly one event is in flight inside the handler. + await subscriber.EmitAsync(itemHandle: 7, value: 0); + await handlerEntered.Task.WaitAsync(Budget); + + // Stage 2 — the rest. `channelCapacity` are accepted; the remainder must be dropped. + for (var i = 1; i < totalEvents; i++) { await subscriber.EmitAsync(itemHandle: 7, value: i); } - // Give the producer a beat to run TryWrite for every event. - await Task.Delay(150); - // Capacity 2 + 1 in-flight in the dispatcher = 3 may have been accepted; the - // remainder should have hit the dropped counter. Don't pin exact counts — - // the scheduler can interleave; pin the invariants instead. - counters.Received.ShouldBeGreaterThanOrEqualTo(totalEvents); - counters.Dropped.ShouldBeGreaterThan(0, - "with capacity=2 and a held dispatcher we must drop at least one of 10 events"); - (counters.Received).ShouldBe(counters.Dispatched + counters.Dropped + counters.InFlight, - "received = dispatched + dropped + (events still queued)"); + // PRESENCE assertion on the discriminator (#500's rule): poll until the drop count reaches its + // terminal value rather than sleeping a fixed 150 ms and hoping the producer got scheduled. + // The budget is an upper bound before giving up — spending it costs nothing on the happy path, + // and no amount of it can make a genuinely broken pump pass. Dropped starts at 0, so this + // cannot be satisfied by the initial state. + await WaitUntilAsync(() => counters.Dropped == expectedDropped, + $"expected {expectedDropped} dropped events"); - // Release the dispatcher so DisposeAsync can drain cleanly. - dispatchGate.TrySetResult(); + counters.Received.ShouldBe(totalEvents, + "the producer must keep reading the gw stream through the overflow, not back-pressure"); + counters.Dropped.ShouldBe(expectedDropped); + counters.Dispatched.ShouldBe(0, + "the dispatch loop is still blocked in the handler, so nothing has completed dispatch — " + + "if this is non-zero the handler is not actually holding the loop and the test proves nothing"); } finally { + // Release BEFORE disposing: DisposeAsync awaits the dispatch loop, and that loop is parked on + // this gate. Releasing it here (not at the end of the try) is what keeps a failed assertion a + // failure instead of a hang. + dispatchGate.Set(); + if (pump is not null) await pump.DisposeAsync(); + dispatchGate.Dispose(); counters.Dispose(); } } @@ -110,23 +160,21 @@ public sealed class EventPumpBoundedChannelTests // Poll until at least one galaxy.events.received measurement tagged // galaxy.client=Driver-X lands in the listener, rather than using a // fixed delay that races under parallel test load on a busy box. - var deadline = DateTime.UtcNow.AddSeconds(5); - bool found = false; - while (DateTime.UtcNow < deadline) - { - listener.RecordObservableInstruments(); - bool hasMatch; - lock (captured) + // Budget, not 5 s: the same contention that produced #503 stretches how long the + // producer loop takes to be scheduled, and a presence poll costs nothing to over-budget. + await WaitUntilAsync( + () => { - hasMatch = captured.Any(c => - c.Instrument == "galaxy.events.received" && - c.Tags.Any(t => t.Key == "galaxy.client" && - string.Equals((string?)t.Value, "Driver-X", StringComparison.Ordinal))); - } - if (hasMatch) { found = true; break; } - await Task.Delay(25); - } - _ = found; // assertion happens below after dispose + listener.RecordObservableInstruments(); + lock (captured) + { + return captured.Any(c => + c.Instrument == "galaxy.events.received" && + c.Tags.Any(t => t.Key == "galaxy.client" && + string.Equals((string?)t.Value, "Driver-X", StringComparison.Ordinal))); + } + }, + "a galaxy.events.received measurement tagged galaxy.client=Driver-X"); } // The static Meter is shared across all EventPump instances in the test @@ -146,7 +194,46 @@ public sealed class EventPumpBoundedChannelTests ours.ShouldContain(c => c.Instrument == "galaxy.events.received"); } - private static CounterCapture StartMeterCapture() + /// Polls until it holds or elapses, failing + /// with on timeout. The presence-assertion counterpart to a fixed delay. + private static async Task WaitUntilAsync(Func condition, string what) + { + var deadline = DateTime.UtcNow + Budget; + while (DateTime.UtcNow < deadline) + { + if (condition()) return; + await Task.Delay(10); + } + condition().ShouldBeTrue($"timed out after {Budget.TotalSeconds:0}s waiting: {what}"); + } + + /// + /// Captures the three EventPump counters for one specific pump, identified by its + /// galaxy.client tag. + /// + /// + /// + /// The galaxy.client filter is the fix for #503, and it is not optional. The pump's + /// is static — one instance for the whole process — and + /// InstrumentPublished can only select by meter, so an unfiltered listener + /// receives measurements from every alive anywhere in the assembly. + /// xUnit runs test classes in parallel and several of them build pumps + /// (EventPumpStreamFaultTests, the Driver-X test below), so a sibling's events + /// landed in these counters — that is why the flake needed a busy box: it needed the two + /// tests to overlap in time. The leak was measured directly, as Received = 11 in a run + /// that emitted 10. + /// + /// + /// The old Received == Dispatched + Dropped + InFlight assertion is what turned the + /// leak into a failure: a foreign increment landing between its four separate + /// calls made the two sides disagree. Hence the + /// reported subject, counters.Received. + /// + /// + /// The galaxy.client tag value to accept. Must be unique to the + /// calling test — pass a fresh GUID-suffixed name rather than a literal, so a future sibling test + /// cannot silently reintroduce the leak by reusing the same name. + private static CounterCapture StartMeterCapture(string clientName) { var capture = new CounterCapture(); var listener = new MeterListener(); @@ -154,8 +241,20 @@ public sealed class EventPumpBoundedChannelTests { if (instr.Meter.Name == EventPump.MeterName) l.EnableMeasurementEvents(instr); }; - listener.SetMeasurementEventCallback((instr, value, _, _) => + listener.SetMeasurementEventCallback((instr, value, tags, _) => { + var mine = false; + foreach (var tag in tags) + { + if (tag.Key == "galaxy.client" && + string.Equals((string?)tag.Value, clientName, StringComparison.Ordinal)) + { + mine = true; + break; + } + } + if (!mine) return; + switch (instr.Name) { case "galaxy.events.received": Interlocked.Add(ref capture._received, value); break; @@ -178,8 +277,11 @@ public sealed class EventPumpBoundedChannelTests public long Dispatched => Interlocked.Read(ref _dispatched); /// Gets the count of dropped events. public long Dropped => Interlocked.Read(ref _dropped); - /// Gets the count of in-flight events. - public long InFlight => Math.Max(0, Received - Dispatched - Dropped); + // There is deliberately no InFlight member. It could only be DERIVED as + // Received - Dispatched - Dropped, which made the old + // `Received == Dispatched + Dropped + InFlight` assertion a tautology: it could not fail on a + // real defect, only on a torn read across the four separate Interlocked.Read calls. Assert the + // three measured counters directly instead (#503). /// Disposes the meter listener. public void Dispose() => Listener?.Dispose(); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmAckRoutingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmAckRoutingTests.cs index e951bb60..65fc6e57 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmAckRoutingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmAckRoutingTests.cs @@ -86,8 +86,19 @@ public sealed class DriverHostActorNativeAlarmAckRoutingTests : RuntimeActorTest actor.Tell(new DriverHostActor.RouteNativeAlarmAck("Plant/gw/dev1/does-not-exist", "cmt", "alice")); - // Give the (fire-and-forget) handler time to run; the unmapped node must produce no ack. - AwaitAssert(() => recorder.Acks.ShouldBeEmpty(), duration: TimeSpan.FromMilliseconds(800)); + // POSITIVE CONTROL, not a sleep (#501). `Acks` is empty at t=0, so any assertion made before the + // host has processed the message above passes without testing anything — and AwaitAssert returns on + // its FIRST success, so it spends none of its duration and is the wrong tool for an absence. + // + // Instead send a MAPPED ack afterwards. The host processes its mailbox in order and forwards to the + // child via Tell, and the child handles RouteAlarmAck with ReceiveAsync (which suspends its own + // mailbox), so by the time this one reaches the driver the unmapped one has provably already been + // handled. Whatever `Acks` holds then is the complete answer. + actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "control", "alice")); + + AwaitAssert(() => recorder.Acks.ShouldContain(a => a.Comment == "control"), duration: Timeout); + recorder.Acks.Count.ShouldBe(1, + "only the control ack may reach the driver — the unmapped condition NodeId must have been dropped"); } /// On a SECONDARY node the ack is gated off (same primary gate as the inbound write path): the @@ -110,10 +121,25 @@ public sealed class DriverHostActorNativeAlarmAckRoutingTests : RuntimeActorTest }, CorrelationId.NewId())); - actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "cmt", "alice")); + actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "gated", "alice")); - // No ack reached the driver — the gate short-circuited before the inverse-map lookup. - AwaitAssert(() => recorder.Acks.ShouldBeEmpty(), duration: TimeSpan.FromMilliseconds(800)); + // POSITIVE CONTROL (#501) — same reasoning as the unmapped-node test, with the role itself as the + // control: return the node to PRIMARY and re-send the SAME mapped ack. Mailbox ordering means that + // once this one reaches the driver, the Secondary-gated one has already been processed. The + // discriminator is the comment, so a leaked ack is visible rather than merged into the count. + actor.Tell(new RedundancyStateChanged( + new[] + { + new NodeRedundancyState(TestNode, RedundancyRole.Primary, + IsClusterLeader: true, IsDriverPrimary: true, AsOfUtc: DateTime.UtcNow), + }, + CorrelationId.NewId())); + actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "control", "alice")); + + AwaitAssert(() => recorder.Acks.ShouldContain(a => a.Comment == "control"), duration: Timeout); + recorder.Acks.ShouldNotContain(a => a.Comment == "gated", + "the Primary gate must have dropped the ack sent while this node was Secondary"); + recorder.Acks.Count.ShouldBe(1); } /// Spawns the host with the recording alarm-driver factory, dispatches the deployment, and waits diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs new file mode 100644 index 00000000..1a901650 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorSubscriptionReconcileTests.cs @@ -0,0 +1,122 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Gitea #488 — the periodic desired-vs-actual subscription reconcile that rides +/// 's existing health-poll tick. +/// +/// +/// +/// The hole it closes: the desired ref set is re-applied on a Connected transition and on a +/// deploy's SetDesiredSubscriptions, and nowhere else. A subscription lost while the driver +/// stayed Connected — the shape reproduced here, a subscribe that failed and was never +/// retried — left the actor permanently subscribed to nothing while reporting Healthy. +/// +/// +/// All three tests drive a short healthPollInterval rather than the production 30 s. The +/// two absence assertions are bounded by a positive control (a counted number of ticks provably +/// processed), never by a bare sleep. +/// +/// +public sealed class DriverInstanceActorSubscriptionReconcileTests : RuntimeActorTestBase +{ + private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(100); + + /// + /// The reconcile itself: the first subscribe fails, leaving the actor Connected with a desired set + /// and no handle. Nothing else would ever retry it — no reconnect, no deploy — so a second + /// subscribe can only come from the periodic reconcile. + /// + [Fact] + public void A_lost_subscription_is_re_established_while_still_connected() + { + var driver = new SubscribableStubDriver { SubscribeFailuresBeforeSuccess = 1 }; + var actor = Sys.ActorOf(DriverInstanceActor.Props(driver, healthPollInterval: FastPoll)); + + actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( + new[] { "tag-a" }, TimeSpan.FromMilliseconds(100))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + // The connect-path subscribe runs and throws, so the handle stays null. + AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget); + + // Only the reconcile can produce this second attempt — and it must succeed this time. + AwaitCondition(() => driver.SubscribeCount >= 2, PresenceBudget); + driver.LastSubscribedRefs.ShouldBe(new[] { "tag-a" }); + } + + /// + /// A healthy subscription is left alone — the reconcile must not re-subscribe on every tick, which + /// would churn the backend every 30 s in production for no reason. + /// + [Fact] + public void A_healthy_subscription_is_not_re_subscribed_on_every_tick() + { + var publisher = new CountingHealthPublisher(); + var driver = new SubscribableStubDriver(); + var actor = Sys.ActorOf(DriverInstanceActor.Props( + driver, healthPublisher: publisher, healthPollInterval: FastPoll)); + + actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( + new[] { "tag-a" }, TimeSpan.FromMilliseconds(100))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget); + + // POSITIVE CONTROL for the absence below. Every health-poll tick publishes a snapshot, so the + // publisher count is a direct observation of ticks PROCESSED. Waiting for several of them is what + // makes "SubscribeCount is still 1" mean "the reconcile ran and declined" rather than "the timer + // had not fired yet" — the latter would pass instantly and prove nothing. + var before = publisher.Count; + AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget); + + driver.SubscribeCount.ShouldBe(1, + "the handle is live, so the reconcile must leave it alone"); + } + + /// + /// A driver that is not can still be handed a desired set. Reconciling + /// it would tell Subscribe every tick and fail every tick, forever — so the reconcile is + /// gated on the capability. + /// + [Fact] + public void A_non_subscribable_driver_is_never_reconciled() + { + var publisher = new CountingHealthPublisher(); + var driver = new StubDriver(); // IDriver only — no ISubscribable + var actor = Sys.ActorOf(DriverInstanceActor.Props( + driver, healthPublisher: publisher, healthPollInterval: FastPoll)); + + actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( + new[] { "tag-a" }, TimeSpan.FromMilliseconds(100))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + // Same positive control: assert the reconcile stays silent across several PROCESSED ticks. + // EventFilter bounds the absence to this block, and the reconcile announces itself at Info — + // the first test above is the proof that this message does appear when the reconcile fires. + EventFilter.Info(contains: "periodic reconcile").Expect(0, () => + { + var before = publisher.Count; + AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget); + }); + } + + /// Counts health-snapshot publishes, which is one per health-poll tick once the actor + /// is running — the observable that turns "time passed" into "ticks were processed". + private sealed class CountingHealthPublisher : IDriverHealthPublisher + { + private int _count; + + /// Number of snapshots published so far. + public int Count => Volatile.Read(ref _count); + + /// + public void Publish(string clusterId, string driverInstanceId, DriverHealth health, int errorCount5Min) + => Interlocked.Increment(ref _count); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs index 9e83fb46..2ab4213d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs @@ -114,6 +114,18 @@ internal sealed class SubscribableStubDriver : StubDriver, ISubscribable /// synchronously-completed stub task hides (the continuation otherwise runs inline). public bool UnsubscribeYields { get; set; } + /// + /// How many of the first calls throw before one succeeds. Default 0 + /// (always succeed) leaves existing behaviour untouched. + /// + /// + /// This is how a test reaches the state the #488 reconcile exists for: the actor is Connected and + /// holds a desired ref set, but the subscribe that should have established the handle failed, so + /// _subscriptionHandle is null and nothing else will ever re-assert it — there is no + /// connect transition coming, and no deploy. + /// + public int SubscribeFailuresBeforeSuccess { get; set; } + /// Subscribes to the specified full references. /// The full references to subscribe to. /// The publishing interval. @@ -121,8 +133,13 @@ internal sealed class SubscribableStubDriver : StubDriver, ISubscribable public Task SubscribeAsync( IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) { - Interlocked.Increment(ref SubscribeCount); + var attempt = Interlocked.Increment(ref SubscribeCount); LastSubscribedRefs = fullReferences; + if (attempt <= SubscribeFailuresBeforeSuccess) + { + throw new InvalidOperationException($"stub subscribe failure #{attempt}"); + } + return Task.FromResult(_handle); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/PeerProbeSupervisorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/PeerProbeSupervisorTests.cs index 5937cd3e..46c2987d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/PeerProbeSupervisorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/PeerProbeSupervisorTests.cs @@ -78,7 +78,9 @@ public sealed class PeerProbeSupervisorTests : RuntimeActorTestBase duration: PresenceBudget); } - /// Verifies a single-node snapshot (just the local node) spawns no children. + /// Verifies a single-node snapshot (just the local node) spawns no children — the local node + /// is never its own probe peer. Established against a following two-node snapshot as a positive + /// control, so the count of 1 (rather than 2) is what carries the "no local child" claim. [Fact] public void Single_node_snapshot_spawns_no_children() { @@ -87,7 +89,20 @@ public sealed class PeerProbeSupervisorTests : RuntimeActorTestBase sup.Tell(Snapshot(State(Local, RedundancyRole.Primary))); - AwaitAssert(() => sup.UnderlyingActor.ChildCount.ShouldBe(0), + // POSITIVE CONTROL (#501). This is an ABSENCE assertion, and ChildCount is 0 before the snapshot + // above has been processed at all — so asserting it directly passes at t=0 and proves nothing + // (AwaitAssert returns on its first success, spending none of its duration). Unlike the other + // ChildCount.ShouldBe(0) waits in this class, there is no preceding ShouldBe(1) to make it a + // transition. + // + // Follow with a snapshot that DOES spawn exactly one child. The supervisor processes its mailbox + // in order, so once the count reaches 1 the single-node snapshot has provably been handled — and + // had it spawned a child for the local node, the count would be 2. + sup.Tell(Snapshot( + State(Local, RedundancyRole.Primary), + State(Peer, RedundancyRole.Secondary))); + + AwaitAssert(() => sup.UnderlyingActor.ChildCount.ShouldBe(1), duration: PresenceBudget); }