fix(historian): retention applock release hardening, shutdown-quiet cancellation, release tests, docs (#667 review)

docs/Historian.md

@@ -151,7 +151,7 @@ and all HistoryRead calls on historized nodes return `GoodNoData` (empty, not an
 | `SqlServer:Schema` | string | `hist` | Must match `^[A-Za-z_][A-Za-z0-9_]{0,127}$` (it is spliced into DDL/DML) | SQL schema that holds `SchemaVersion`, `Tag`, `Sample`, `Event` and the `SampleRow`/`TimeRow` table types. |
 | `SqlServer:CommandTimeout` | TimeSpan | `00:00:30` | Must be `> 0` | Per-command timeout, rounded up to whole seconds (minimum 1 s). The SQL backend does **not** use `CallTimeout`. |
 | `SqlServer:AutoCreateSchema` | bool | `true` | — | `true`: on first use apply any missing embedded script under a `sp_getapplock` (`OtOpcUaHistorianSchema`), so concurrent nodes serialise. `false`: only verify `SchemaVersion >= 1`; if behind, **every** historian call fails with a clear error naming the missing version (surfaced in the read client's health `LastError`, never a host crash). A failed bootstrap is retried on the next call, not latched. |
-| `SqlServer:RetentionDays` | int? | unset | `>= 1` when set | Days of history kept, on **source time**. Unset keeps everything. When set, the Primary driver node purges older `hist.Sample` / `hist.Event` rows — see [Retention](#retention-retentiondays). |
+| `SqlServer:RetentionDays` | int? | unset | `>= 1` when set | Days of history kept, on **source time**. Unset keeps everything. When set, driver nodes purge older `hist.Sample` / `hist.Event` rows — see [Retention](#retention-retentiondays). |
 | `SqlServer:PurgeInterval` | TimeSpan | `01:00:00` | `> 0` | How often the purge runs (first pass at start-up). |
 | `SqlServer:PurgeBatchSize` | int | `4000` | `>= 1` | Rows per `DELETE TOP (n)`; the default stays under SQL Server's ~5000-lock escalation threshold so a purge never takes a table lock. |
 
@@ -782,8 +782,11 @@ equivalent TTL (`QuestDb:RetentionDays`); the Gateway backend's retention is the
 
 - **What runs.** `HistorianRetentionService` (a hosted service, driver-role nodes, Backend=`SqlServer` with `RetentionDays` set; inert
   otherwise) ticks every `PurgeInterval` (first pass at start-up) and calls the backend's `IHistorianRetention.PurgeAsync`.
-- **Primary-gated.** Each tick consults `IHistorizationRoleView.ShouldRecordHistorizedValues` (the same verdict that gates capture); a
-  standby skips, an unknown role runs.
+- **Gate vs. mutual exclusion.** Each tick consults `IHistorizationRoleView.ShouldRecordHistorizedValues` (the capture verdict; a
+  standby skips, an unknown role runs). That verdict is **pair-scoped** while the purge is **database-scoped**, so the gate is only a
+  load reducer: the `sp_getapplock` below is the actual mutual exclusion. Pairs sharing one historian database, and both nodes of a
+  pair on their boot-time pass, simply serialise on it. Set `RetentionDays` **uniformly on every node that shares a historian
+  database** — if they differ, the smallest window wins.
 - **How it deletes.** Cutoff = `UtcNow - RetentionDays` on source time. Samples are purged **per tag** with
   `DELETE TOP (@n) ... WHERE TagId = @id AND SourceTimeUtc < @cutoff` (a seek on the clustered `(TagId, SourceTimeUtc)` key; a global
   time predicate would scan), then events with `DELETE TOP (@n) ... WHERE EventTimeUtc < @cutoff` (seek on `CX_Event_Time`), each looped

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs

@@ -36,20 +36,23 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention
 
     public TimeSpan Interval { get; }
 
+    /// <summary>Test hook invoked after every delete batch.</summary>
+    internal Action? AfterBatch { get; set; }
+
     public async Task<IReadOnlyList<RetentionPurgeResult>> PurgeAsync(CancellationToken ct)
     {
         await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
         var cutoff = _time.GetUtcNow().UtcDateTime.AddDays(-_days);
 
         await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
-        if (!await TryLockAsync(conn, ct).ConfigureAwait(false))
-        {
-            _log.LogDebug("Historian retention purge skipped: another purger holds the applock.");
-            return [];
-        }
-
         try
         {
+            if (!await TryLockAsync(conn, ct).ConfigureAwait(false))
+            {
+                _log.LogInformation("Historian retention purge skipped: another purger holds the applock.");
+                return [];
+            }
+
             long samples = 0;
             var tagIds = new List<int>();
             await using (var cmd = _conn.CreateCommand(conn, $"SELECT TagId FROM {_conn.Q("Tag")}"))
@@ -71,6 +74,7 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention
         }
         finally
         {
+            // Also runs when acquire threw: the server may have granted the lock before the client saw the reply.
             await ReleaseAsync(conn).ConfigureAwait(false);
         }
     }
@@ -87,6 +91,7 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention
             SqlHistorianConnection.AddDateTime2(cmd, "@cutoff", cutoff);
             var n = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
             total += n;
+            AfterBatch?.Invoke();
             if (n < _batch) return total;
         }
     }
@@ -112,7 +117,11 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention
         }
         catch (Exception ex)
         {
-            _log.LogDebug(ex, "Releasing the retention applock failed.");
+            // A pooled connection keeps its server session (sp_reset_connection does not release a Session-owned
+            // applock); clearing the pool makes the dispose physically close it, which releases the lock.
+            _log.LogWarning("sp_releaseapplock failed ({Exception}); discarding the connection pool so the session " +
+                "holding the retention applock is closed rather than pooled.", ex.GetType().Name);
+            SqlConnection.ClearPool(conn);
         }
     }
 }

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianRetentionService.cs

@@ -66,8 +66,9 @@ public sealed class HistorianRetentionService : BackgroundService
             }
             return true;
         }
-        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        catch (Exception) when (ct.IsCancellationRequested)
         {
+            // Shutdown: SqlClient surfaces cancellation as SqlException, not only OperationCanceledException.
             return false;
         }
         catch (Exception ex)

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs

@@ -119,15 +119,56 @@ public sealed class RetentionTests(SqlHistorianFixture fx)
         (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(6);
     }
 
+    /// <summary>True when a fresh, non-pooled session can take the retention applock (i.e. nobody holds it).</summary>
+    private async Task<bool> LockIsFreeAsync(string s)
+    {
+        var cs = new SqlConnectionStringBuilder(fx.ConnectionString) { Pooling = false }.ConnectionString;
+        await using var c = new SqlConnection(cs);
+        await c.OpenAsync();
+        await using var cmd = c.CreateCommand();
+        cmd.CommandText = $"DECLARE @r int; EXEC @r = sp_getapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT @r;";
+        return Convert.ToInt32(await cmd.ExecuteScalarAsync()) >= 0;
+    }
+
     [Fact]
-    public async Task Cancellation_stops_the_purge()
+    public async Task The_applock_is_released_after_a_purge()
+    {
+        fx.SkipIfUnavailable();
+        var (c, b, _, s) = Make();
+        await SeedAsync(b, s);
+        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now));
+        await purger.PurgeAsync(CancellationToken.None);
+        // A pooled-session re-entry would hide a missing release; a separate non-pooled session cannot.
+        (await LockIsFreeAsync(s)).ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task Cancelling_between_batches_stops_the_purge_and_still_releases_the_applock()
+    {
+        fx.SkipIfUnavailable();
+        var (c, b, _, s) = Make();
+        await SeedAsync(b, s);
+        using var cts = new CancellationTokenSource();
+        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 1, time: new FixedTime(Now));
+        var batches = 0;
+        purger.AfterBatch = () => { if (++batches == 2) cts.Cancel(); };
+
+        await Should.ThrowAsync<OperationCanceledException>(() => purger.PurgeAsync(cts.Token));
+
+        batches.ShouldBe(2);
+        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(19); // 21 - two batches of 1
+        (await LockIsFreeAsync(s)).ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task A_throwing_purge_releases_the_applock()
     {
         fx.SkipIfUnavailable();
         var (c, b, _, s) = Make();
         await SeedAsync(b, s);
         var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 1, time: new FixedTime(Now));
-        using var cts = new CancellationTokenSource();
-        cts.Cancel();
-        await Should.ThrowAsync<OperationCanceledException>(() => purger.PurgeAsync(cts.Token));
+        purger.AfterBatch = () => throw new InvalidOperationException("boom");
+        await Should.ThrowAsync<InvalidOperationException>(() => purger.PurgeAsync(CancellationToken.None));
+        (await LockIsFreeAsync(s)).ShouldBeTrue();
     }
 }