diff --git a/docs/Historian.md b/docs/Historian.md index 643220ca..7d30c0e2 100644 --- a/docs/Historian.md +++ b/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 diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs index a729495e..3a39dd9e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs @@ -36,20 +36,23 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention public TimeSpan Interval { get; } + /// Test hook invoked after every delete batch. + internal Action? AfterBatch { get; set; } + public async Task> 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(); 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); } } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianRetentionService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianRetentionService.cs index 52f3edf1..aaa8d826 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianRetentionService.cs +++ b/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) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs index 4dc56b97..85960bd4 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs +++ b/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); } + /// True when a fresh, non-pooled session can take the retention applock (i.e. nobody holds it). + private async Task 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(() => 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(() => purger.PurgeAsync(cts.Token)); + purger.AfterBatch = () => throw new InvalidOperationException("boom"); + await Should.ThrowAsync(() => purger.PurgeAsync(CancellationToken.None)); + (await LockIsFreeAsync(s)).ShouldBeTrue(); } }