Merge #602 stability: FOCAS/AbLegacy loop joins on shutdown, Postgres untyped string binds, MTConnect pump generation, MQTT birth-typed writes + QoS1 plain writes + claim backoff

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 07:33:03 -04:00
parent 85491f1ebc babc452740
commit b29642c5da
21 changed files with 691 additions and 31 deletions
@@ -0,0 +1,54 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.PlcFamilies;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests;
+
+/// <summary>
+///     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.
+/// </summary>
+[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
+        }
+    }
+
+    /// <summary>Shutdown blocks on a probe loop parked in runtime init, then the loop disposes its probe runtime.</summary>
+    [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();
+    }
+}
@@ -0,0 +1,57 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Driver.FOCAS;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests;
+
+/// <summary>
+///     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.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class FocasShutdownJoinTests
+{
+    /// <summary>A client whose connect parks until released and deliberately ignores the cancellation token.</summary>
+    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);
+        }
+    }
+
+    /// <summary>Shutdown blocks on a probe loop parked in connect, then disposes the client that connect produced.</summary>
+    [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);
+    }
+}
@@ -0,0 +1,98 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
+
+/// <summary>
+///     STAB-20 / #602 — a Subscribe that lands between <c>StopSampleStreamAsync</c> 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.
+/// </summary>
+public sealed class MTConnectPumpGenerationTests
+{
+    private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(10);
+
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    /// <summary>The pump a mid-restart Subscribe started on the retired client is replaced by one on the new client.</summary>
+    [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<CannedAgentClient>([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))]);
+}
@@ -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);
@@ -238,6 +238,53 @@ public sealed class SparkplugPrimaryHostTests
        await host.DisposeAsync();
    }

+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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();
+    }
+
+    /// <summary>The ladder is exponential from 2 s and capped, so a long outage settles at one attempt per cap.</summary>
+    [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;
+    }
+
    /// <summary>A broker that refuses instantly, so a connect attempt is observable but not slow.</summary>
    private static MqttDriverOptions Unreachable() =>
        Options() with { Host = "127.0.0.1", Port = 1, ConnectTimeoutSeconds = 1 };
@@ -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 ───────────────────────────────────────────────────
@@ -114,6 +114,63 @@ public sealed class PostgresReadTests
            .ShouldBe(PostgresFixture.NewestStationOvenTemp);
    }

+    // ---- integer key columns (STAB-24) ----
+
+    /// <summary>
+    ///     A key-value tag whose key column is <c>integer</c> reads. Authored <c>keyValue</c> is text, and
+    ///     Npgsql's default <c>text</c> bind has no <c>integer = text</c> operator — the read failed with a
+    ///     provider error before <c>PostgresDialect.ConfigureParameter</c> sent string binds untyped.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>A wide-row <c>whereColumn</c> selector over the <c>integer station_id</c> column reads the matching row.</summary>
+    [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) ----

    /// <summary>
@@ -104,6 +104,16 @@ public sealed class ColonMarkerDialect : ISqlDialect
        lock (_commands) _commands.Add(command);
    }

+    /// <summary>Every value handed to <see cref="ConfigureParameter"/>, in call order.</summary>
+    public List<object?> ConfiguredParameterValues { get; } = [];
+
+    /// <inheritdoc/>
+    public void ConfigureParameter(DbParameter parameter, object? value)
+    {
+        ArgumentNullException.ThrowIfNull(parameter);
+        lock (ConfiguredParameterValues) ConfiguredParameterValues.Add(value);
+    }
+
    /// <inheritdoc/>
    public string QuoteIdentifier(string ident) => Inner.QuoteIdentifier(ident);

@@ -24,6 +24,26 @@ public class PostgresDialectTests
        d.QuoteIdentifier("a\"b").ShouldBe("\"a\"\"b\"");
    }

+    /// <summary>STAB-24: a string bind is untyped so Postgres infers it against an integer key column.</summary>
+    [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);
+    }
+
+    /// <summary>Non-string values keep their natural type, and a foreign parameter type is tolerated.</summary>
+    [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<ArgumentException>(() => new PostgresDialect().QuoteIdentifier("x\0y"));
@@ -318,6 +318,35 @@ public sealed class SqlCatalogGateDriverTests
        dialect.ConfiguredCommands.ShouldBeGreaterThan(0);
    }

+    /// <summary>
+    ///     STAB-24: every poll-path bind goes through <see cref="ISqlDialect.ConfigureParameter"/> with the
+    ///     authored value, so a dialect can type it. Without the call the Postgres untyped-bind fix is inert.
+    /// </summary>
+    [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<SqlDriver>.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 ----

    /// <summary>