Merge review R2 fixes: Galaxy subscribe-once deadline returns BadTimeout, S7 probe honours backoff, write success clears only write-side Faulted, null array elements rejected, S7_SIM_ALLOW_WRITES opt-in

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe

docs/drivers/S7-Test-Fixture.md

@@ -20,6 +20,8 @@ python-snap7 `Server` via `Docker/docker-compose.yml --profile s7_1500`
 on `10.100.0.35:1102` (the shared Docker host; override via `S7_SIM_ENDPOINT`;
 pinned `python:3.12-slim-bookworm` base +
 `python-snap7>=2.0`). Docker is the only supported launch path.
+Tests that WRITE to the endpoint (currently `S7_1500ArrayWriteTests`) additionally require
+`S7_SIM_ALLOW_WRITES=1`, because `S7_SIM_ENDPOINT` can be pointed at a real CPU; they skip without it.
 `Snap7ServerFixture` probes the port at collection init + skips with a
 clear message when unreachable (matches the pymodbus pattern).
 `server.py` (baked into the image under `Docker/`) reads a JSON profile

docs/drivers/S7.md

@@ -252,7 +252,9 @@ is written as ONE contiguous `WriteBytesAsync` of `count x element-bytes` (the i
 decode), and the value must be an array of exactly `arrayLength` elements (else `BadTypeMismatch`, no
 health impact). **Writable `Bool` arrays are refused at init** (`NotSupportedException`): the array
 read is byte-granular, so a matching write would overwrite the 7 neighbouring bits of every byte it
-touches, and a safe bit-packed write needs read-modify-write. Mark Bool arrays `writable: false`.
+touches, and a safe bit-packed write needs read-modify-write. Mark Bool arrays `writable: false`. A `null` array element is rejected (`BadTypeMismatch`, nothing written)
+rather than coerced to 0. An array write larger than one PDU is split by S7.Net into several writes and is
+**not atomic** — a mid-write failure can leave the PLC array partly updated.
 Note the *server* still presents every array node read-only to OPC UA clients ([Uns.md §Array
 tags](../Uns.md#array-tags-1-d), #610 item 1) — the driver path is reachable today through the Client
 CLI / driver-level callers and is proven by the unit suite plus the snap7 fixture test

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs

@@ -812,6 +812,9 @@ public sealed class GalaxyDriver
         return ReadViaSubscribeOnceAsync(dialledRefs, cancellationToken);
     }
 
+    /// <summary>Upper bound on one subscribe-once read; internal so tests can shrink it.</summary>
+    internal TimeSpan ReadViaSubscribeDeadline { get; set; } = TimeSpan.FromSeconds(30);
+
     /// <summary>
     ///     Production read path. MxAccess has no one-shot Read RPC — every value comes
     ///     through the event stream. We synthesise a Read by:
@@ -826,9 +829,6 @@ public sealed class GalaxyDriver
     ///     caller's cancellation token fires, return a Bad-status snapshot in input order
     ///     so the caller still sees one snapshot per requested reference.
     /// </summary>
-    /// <summary>Upper bound on one subscribe-once read; internal so tests can shrink it.</summary>
-    internal TimeSpan ReadViaSubscribeDeadline { get; set; } = TimeSpan.FromSeconds(30);
-
     private async Task<IReadOnlyList<DataValueSnapshot>> ReadViaSubscribeOnceAsync(
         IReadOnlyList<string> fullReferences, CancellationToken callerToken)
     {
@@ -868,11 +868,36 @@ public sealed class GalaxyDriver
 
         var bufferedIntervalMs = _options.MxAccess.PublishingIntervalMs;
         IReadOnlyList<SubscribeResult> results;
+        // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow
+        // gateway can never surface an OperationCanceledException to a caller that did not cancel, and the
+        // handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call.
+        var subscribeTask = _subscriber!.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken);
         try
         {
-            results = await _subscriber!
-                .SubscribeBulkAsync(fullReferences, bufferedIntervalMs, cancellationToken)
-                .ConfigureAwait(false);
+            results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (!callerToken.IsCancellationRequested)
+        {
+            // Internal deadline fired while the subscribe RPC was still in flight.
+            pump.OnDataChange -= handler;
+            var subscriber = _subscriber!;
+            var logger = _logger;
+            _ = subscribeTask.ContinueWith(t =>
+            {
+                if (!t.IsCompletedSuccessfully) return Task.CompletedTask;
+                var late = t.Result.Where(r => r.WasSuccessful && r.ItemHandle > 0).Select(r => r.ItemHandle).ToArray();
+                return late.Length == 0
+                    ? Task.CompletedTask
+                    : subscriber.UnsubscribeBulkAsync(late, CancellationToken.None).ContinueWith(
+                        u => logger.LogWarning(u.Exception,
+                            "GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length),
+                        TaskContinuationOptions.OnlyOnFaulted);
+            }, TaskScheduler.Default).Unwrap();
+            var now = DateTime.UtcNow;
+            var timedOut = new DataValueSnapshot[fullReferences.Count];
+            for (var i = 0; i < timedOut.Length; i++)
+                timedOut[i] = new DataValueSnapshot(null, StatusCodeMap.BadTimeout, null, now);
+            return timedOut;
         }
         catch
         {

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs

@@ -1869,6 +1869,7 @@ public sealed class S7Driver
         if (!outcome.HasObservations) return;
 
         var worst = outcome.WorstFailure;
+        var priorWriteFault = _writeFault;
         if (worst == DriverState.Faulted)
             _writeFault = (DriverState.Faulted, outcome.WorstError!);
         else if (outcome.AnySuccess)
@@ -1878,7 +1879,10 @@ public sealed class S7Driver
         {
             // Success-only batch: no health write, except to retire a Faulted that WE (the write
             // side) were holding — a clean write is the only evidence the alert has cleared.
-            if (outcome.AnySuccess && _writeFault is null && Health.State == DriverState.Faulted)
+            // Only a Faulted the write side owns: a read-side Faulted (e.g. access denied on another DB) is the
+            // read pass's to clear, and Health still carries the write fault's own message iff it is ours.
+            if (outcome.AnySuccess && _writeFault is null && priorWriteFault is { } owned
+                && Health.State == DriverState.Faulted && Health.LastError == owned.Message)
                 Health = new DriverHealth(DriverState.Healthy, Health.LastSuccessfulRead, null);
             return;
         }
@@ -2004,6 +2008,9 @@ public sealed class S7Driver
 
         var elements = new List<object?>(count);
         foreach (var e in items) elements.Add(e);
+        // Convert.ToXxx(null) is 0 — a null element must not silently write zero to the PLC.
+        if (elements.Contains(null))
+            throw new S7WriteValueException($"S7 array tag '{tag.Name}': an element is null.");
         if (elements.Count != count)
             throw new S7WriteValueException(
                 $"S7 array tag '{tag.Name}' has {count} elements but the write supplied {elements.Count}.");
@@ -2361,10 +2368,10 @@ public sealed class S7Driver
                 // Probe via S7.Net's low-cost GetCpuStatus — returns the CPU state (Run/Stop)
                 // and is intentionally light on the comms mailbox. Single-word Plc.ReadAsync
                 // would also work but GetCpuStatus doubles as a "PLC actually up" check.
-                using var probeCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
-                probeCts.CancelAfter(_options.Probe.Timeout);
+                using var gateCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+                gateCts.CancelAfter(_options.Probe.Timeout);
 
-                await _gate.WaitAsync(probeCts.Token).ConfigureAwait(false);
+                await _gate.WaitAsync(gateCts.Token).ConfigureAwait(false);
                 try
                 {
                     try
@@ -2372,7 +2379,14 @@ public sealed class S7Driver
                         // EnsureConnectedAsync doubles as the reconnect backstop: if a read/write
                         // marked the handle dead (or the socket dropped between ticks), the probe
                         // reopens it here so recovery doesn't wait for the next data call.
-                        var plc = await EnsureConnectedAsync(probeCts.Token).ConfigureAwait(false);
+                        // It runs on the LOOP token, not the (shorter) probe deadline: the connect is bounded by
+                        // _options.Timeout inside, and a timeout there must count as a failure so STAB-21's
+                        // backoff window opens — cutting it with the probe deadline looked like a caller cancel,
+                        // recorded nothing, and let a black-holed host cost a connect attempt every probe tick.
+                        var plc = await EnsureConnectedAsync(ct).ConfigureAwait(false);
+                        // The probe deadline bounds only the CPU-status round trip.
+                        using var probeCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+                        probeCts.CancelAfter(_options.Probe.Timeout);
                         await plc.ReadStatusAsync(probeCts.Token).ConfigureAwait(false);
                         success = true;
                     }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs

@@ -1,3 +1,4 @@
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
 using Shouldly;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
@@ -179,4 +180,60 @@ public sealed class GalaxyDriverReadTests
         result[0].StatusCode.ShouldBe(StatusCodeMap.BadTimeout);
         subscriber.UnsubscribedHandles.ShouldContain(subscriber.Map["Silent.Tag"]);
     }
+
+    /// <summary>
+    ///     A subscribe RPC that outlives the internal deadline (caller passed None) must yield BadTimeout for
+    ///     every tag, not an OperationCanceledException, and the handles the late-completing RPC created must
+    ///     still be unsubscribed.
+    /// </summary>
+    [Fact]
+    public async Task ReadAsync_SubscribeOncePath_SlowSubscribeBulk_NoCallerToken_ReturnsBadTimeoutAndUnsubscribesLateHandles()
+    {
+        var subscriber = new SlowSubscribeSubscriber();
+        using var driver = new GalaxyDriver(
+            "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
+        {
+            ReadViaSubscribeDeadline = TimeSpan.FromMilliseconds(200),
+        };
+
+        var result = await driver.ReadAsync(["Slow.A", "Slow.B"], CancellationToken.None)
+            .WaitAsync(TimeSpan.FromSeconds(10));
+
+        result.Count.ShouldBe(2);
+        result.ShouldAllBe(r => r.StatusCode == StatusCodeMap.BadTimeout);
+
+        // The RPC finally completes after the read gave up; its handles must not leak.
+        subscriber.Gate.SetResult();
+        var deadline = DateTime.UtcNow.AddSeconds(10);
+        while (subscriber.UnsubscribedHandles.Count < 2 && DateTime.UtcNow < deadline) await Task.Delay(20);
+        subscriber.UnsubscribedHandles.OrderBy(h => h).ShouldBe([101, 102]);
+    }
+
+    private sealed class SlowSubscribeSubscriber : IGalaxySubscriber
+    {
+        private readonly List<int> _unsubscribed = [];
+        public TaskCompletionSource Gate { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        public IReadOnlyList<int> UnsubscribedHandles { get { lock (_unsubscribed) return _unsubscribed.ToArray(); } }
+
+        public async Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
+            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
+        {
+            await Gate.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
+            var h = 100;
+            return fullReferences.Select(r => new SubscribeResult
+            {
+                TagAddress = r, ItemHandle = ++h, WasSuccessful = true,
+            }).ToList();
+        }
+
+        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
+        {
+            lock (_unsubscribed) _unsubscribed.AddRange(itemHandles);
+            return Task.CompletedTask;
+        }
+
+        public IAsyncEnumerable<ZB.MOM.WW.MxGateway.Contracts.Proto.MxEvent> StreamEventsAsync(CancellationToken cancellationToken)
+            => System.Threading.Channels.Channel.CreateUnbounded<ZB.MOM.WW.MxGateway.Contracts.Proto.MxEvent>()
+                .Reader.ReadAllAsync(cancellationToken);
+    }
 }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/Docker/README.md

@@ -57,6 +57,8 @@ rsync -av --rsync-path="sudo rsync" \
 - Default: `localhost:1102` (non-privileged; sidesteps Windows Firewall
   prompt + Linux's root-required bind on port 102).
 - Override with `S7_SIM_ENDPOINT` to point at a real S7 CPU on `:102`.
+- Tests that write (`S7_1500ArrayWriteTests`) run only with `S7_SIM_ALLOW_WRITES=1`, so a real-CPU
+  endpoint is never written by accident.
 - The driver's S7DriverOptions.Port flows through S7netplus's 5-arg
   `Plc(CpuType, host, port, rack, slot)` ctor so the non-standard port
   works end-to-end.

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/S7_1500/S7_1500ArrayWriteTests.cs

@@ -7,7 +7,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests.S7_1500;
 /// <summary>
 ///     #610 item 2 (S7-016): array write round trip against the python-snap7 S7-1500 profile. Skips
 ///     cleanly when the sim is absent. Uses DB1 bytes 700+ (the profile's 1024-byte DB1 has no seed
-///     there), so the writes cannot disturb another test's seeds.
+///     there), so the writes cannot disturb another test's seeds. Writes also require
+///     <c>S7_SIM_ALLOW_WRITES=1</c>.
 /// </summary>
 [Collection(Snap7ServerCollection.Name)]
 [Trait("Category", "Integration")]
@@ -33,6 +34,9 @@ public sealed class S7_1500ArrayWriteTests(Snap7ServerFixture sim)
     public async Task Array_write_then_read_round_trips_through_real_S7comm()
     {
         if (sim.SkipReason is not null) Assert.Skip(sim.SkipReason);
+        // S7_SIM_ENDPOINT may point at a real CPU; writing needs an explicit opt-in (simulator runs only).
+        if (Environment.GetEnvironmentVariable("S7_SIM_ALLOW_WRITES") != "1")
+            Assert.Skip("Write test: set S7_SIM_ALLOW_WRITES=1 to allow writes to the endpoint under test (simulator only).");
 
         await using var drv = new S7Driver(Options(sim.Host, sim.Port), "s7-arr-write");
         await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7ArrayWriteAndHealthTests.cs

@@ -85,6 +85,31 @@ public sealed class S7ArrayWriteAndHealthTests
         Should.Throw<S7Driver.S7WriteValueException>(() => S7Driver.EncodeArrayBlock(tag, addr, new object[] { 1, "x" })); // format
     }
 
+    /// <summary>
+    ///     A null element must not silently coerce to 0 (Convert.ToXxx(null) == 0) and land on the PLC: it is a
+    ///     client error, and nothing reaches the wire. The valid write in the same batch is the positive control.
+    /// </summary>
+    [Fact]
+    public async Task Write_array_with_a_null_element_is_BadTypeMismatch_and_writes_nothing()
+    {
+        var tag = Arr(S7DataType.Int16, "DB1.DBW0", 2);
+        Should.Throw<S7Driver.S7WriteValueException>(() =>
+            S7Driver.EncodeArrayBlock(tag, S7AddressParser.Parse("DB1.DBW0"), new object?[] { 1, null }));
+
+        var factory = new RecordingFactory();
+        using var drv = new S7Driver(Options(tag), "s7-arr-null", factory);
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+
+        var res = await drv.WriteAsync(
+            [new WriteRequest("Arr", new object?[] { 1, null }), new WriteRequest("Arr", new short[] { 7, 8 })],
+            TestContext.Current.CancellationToken);
+
+        res[0].StatusCode.ShouldBe(BadTypeMismatch);
+        res[1].StatusCode.ShouldBe(0u);
+        factory.Created[0].BlockWrites.ShouldHaveSingleItem().Bytes.ShouldBe(new byte[] { 0, 7, 0, 8 });
+        drv.GetHealth().State.ShouldBe(DriverState.Healthy);
+    }
+
     /// <summary>Bool and wide-type arrays are refused with NotSupported (BadNotSupported on the wire).</summary>
     [Theory]
     [InlineData(S7DataType.Bool, "DB1.DBX0.0")]
@@ -260,6 +285,30 @@ public sealed class S7ArrayWriteAndHealthTests
         drv.GetHealth().State.ShouldBe(DriverState.Healthy);
     }
 
+    /// <summary>
+    ///     A Faulted that a READ pass raised (access denied on a read) is not the write side's to retire: a
+    ///     successful write must leave it in place until a clean read pass clears it.
+    /// </summary>
+    [Fact]
+    public async Task Successful_write_does_not_clear_a_read_side_Faulted()
+    {
+        var (drv, factory) = await TwoTagDriver();
+        using var _ = drv;
+        var plc = factory.Created[0];
+
+        plc.NextReadThrows = new PlcException(ErrorCode.ReadData, "Accessing object not allowed");
+        await drv.ReadAsync(["A"], TestContext.Current.CancellationToken);
+        drv.GetHealth().State.ShouldBe(DriverState.Faulted);
+
+        var res = await drv.WriteAsync([new WriteRequest("B", (ushort)2)], TestContext.Current.CancellationToken);
+        res[0].StatusCode.ShouldBe(0u);                       // the write really got through
+        drv.GetHealth().State.ShouldBe(DriverState.Faulted);  // ... but it does not own this fault
+
+        // Positive control: the read side retires its own fault.
+        (await drv.ReadAsync(["B"], TestContext.Current.CancellationToken))[0].StatusCode.ShouldBe(0u);
+        drv.GetHealth().State.ShouldBe(DriverState.Healthy);
+    }
+
     /// <summary>A transient (Degraded) write failure is still erased by a clean read, as before.</summary>
     [Fact]
     public async Task Degraded_write_failure_is_still_cleared_by_a_clean_read()

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DriverReconnectTests.cs

@@ -442,6 +442,46 @@ public sealed class S7DriverReconnectTests
         drv.GetHostStatuses()[0].State.ShouldBe(HostState.Running);
     }
 
+    /// <summary>
+    ///     STAB-21 vs the probe. On a black-holed host the probe's connect used to be cut by the (shorter) probe
+    ///     deadline, which is not a caller cancel but was treated as one, so NO failure was recorded and every
+    ///     probe tick paid a fresh connect attempt. The connect must run on the connect deadline and a timeout
+    ///     must open the backoff window: far fewer attempts than ticks.
+    /// </summary>
+    [Fact]
+    public async Task Probe_connect_timeout_on_a_black_holed_host_opens_the_backoff_window()
+    {
+        var opts = new S7DriverOptions
+        {
+            Host = "192.0.2.1",
+            Timeout = TimeSpan.FromMilliseconds(400),
+            Probe = new S7ProbeOptions
+            {
+                Enabled = true,
+                Interval = TimeSpan.FromMilliseconds(20),
+                Timeout = TimeSpan.FromMilliseconds(80),
+            },
+            RawTags = S7RawTags.Entries(new S7TagDefinition("W0", "DB1.DBW0", S7DataType.UInt16, Writable: false)),
+        };
+        var factory = new FakeS7PlcFactory();
+        using var drv = new S7Driver(opts, "s7-probe-blackhole", factory);
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+
+        // Kill the live handle (probe fault), then black-hole every reconnect.
+        factory.OpenHangsUntilCancelledCount = int.MaxValue;
+        factory.Created[0].ReadStatusThrows = new System.Net.Sockets.SocketException();
+
+        // First reconnect attempt must have been made and failed (positive control), then observe a window
+        // well inside the 1 s first backoff window.
+        await WaitUntilAsync(() => factory.Created.Count >= 2, TimeSpan.FromSeconds(5));
+        factory.Created.Count.ShouldBeGreaterThanOrEqualTo(2);
+        var atFirstAttempt = factory.Created.Count;
+        await Task.Delay(TimeSpan.FromMilliseconds(900));
+
+        // Without the fix this is ~10+ (one 80 ms-cut attempt per tick); with it the window suppresses them.
+        (factory.Created.Count - atFirstAttempt).ShouldBeLessThanOrEqualTo(1);
+    }
+
     private static S7DriverOptions ProbeEnabledOptions() => new()
     {
         Host = "192.0.2.1",