From 924eff8d3c93bc3421eabc1b6048fc1413d52e8f Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 08:21:30 -0400 Subject: [PATCH 1/2] feat(historian): SQL Server retention purge (RetentionDays, Primary-gated, applock) (#667) --- CLAUDE.md | 2 +- docs/Configuration.md | 3 + docs/Historian.md | 28 +++- .../SqlServerHistorianOptions.cs | 21 +++ .../SqlServerHistorianProvider.cs | 9 ++ .../SqlServerHistorianRetention.cs | 118 ++++++++++++++++ src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 8 ++ .../Historian/HistorianRetentionService.cs | 95 +++++++++++++ .../Historian/IHistorianBackendProvider.cs | 5 + .../Historian/IHistorianRetention.cs | 25 ++++ .../RetentionTests.cs | 133 ++++++++++++++++++ ...SqlServerHistorianRetentionOptionsTests.cs | 62 ++++++++ .../HistorianRetentionServiceTests.cs | 95 +++++++++++++ 13 files changed, 602 insertions(+), 2 deletions(-) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianRetentionService.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianRetention.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianRetentionOptionsTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianRetentionServiceTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index baa899c2..f9d7760a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1341,7 +1341,7 @@ historian tagname, which is the tag's driver `FullName`. ### Pluggable backends, SQL Server and HistoryUpdate (2026-10) - **Selection:** `ServerHistorian:Backend` (`Gateway` default | `SqlServer` | `QuestDb`; unknown fails host start). `HistorianBackendRegistry` → `IHistorianBackendProvider` builds all five seams; the gateway keys are only required for `Gateway`. `SqlServer` uses `ServerHistorian:SqlServer:{ConnectionString (required), Schema=hist, CommandTimeout=30s, AutoCreateSchema=true}`. -- **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; aggregates (13, simple per-sample definitions, not Part 13) are pushed down to T-SQL with bucket math in **100 ns ticks**. Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. +- **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; aggregates (13, simple per-sample definitions, not Part 13) are pushed down to T-SQL with bucket math in **100 ns ticks**. Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. - **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend". - **Real status fix:** the recorder used to stamp every sample Good (`192`); it now carries the driver's status. ⚠️ On the **Gateway** backend this changes nothing stored: the gateway discards quality (C-002), so AVEVA still stores Good. - **Comms loss is historized as an outage marker** (`Empty` + `BadNoCommunication`, one per tag per outage) on SqlServer; dropped (`DroppedUnsupportedKind`) on Gateway. Outbox records are **format v2** — drain or delete `ContinuousHistorization:OutboxPath` before downgrading a node. diff --git a/docs/Configuration.md b/docs/Configuration.md index 4b23cf3c..2298abc1 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -411,6 +411,9 @@ Validated at host start (driver-role nodes) when `ServerHistorian:Enabled` **or* | `ServerHistorian:SqlServer:Schema` | `hist` | Must match `^[A-Za-z_][A-Za-z0-9_]{0,127}$` (spliced into DDL/DML). | | `ServerHistorian:SqlServer:CommandTimeout` | `00:00:30` | Must be `> 0`; rounded up to whole seconds. | | `ServerHistorian:SqlServer:AutoCreateSchema` | `true` | `true` applies the embedded schema script under an app-lock on first use; `false` only verifies `SchemaVersion >= 1` and fails every historian call (health `LastError`, no crash) if behind. | +| `ServerHistorian:SqlServer:RetentionDays` | unset | Days of history kept (source time), `>= 1` when set; unset keeps everything. Enables the Primary-gated purge of `hist.Sample`/`hist.Event` (see [Historian.md](Historian.md#retention-retentiondays)). | +| `ServerHistorian:SqlServer:PurgeInterval` | `01:00:00` | Purge cadence; must be `> 0`. | +| `ServerHistorian:SqlServer:PurgeBatchSize` | `4000` | Rows per `DELETE TOP (n)` batch; `>= 1`. Keep under ~5000 to avoid lock escalation. | #### `ServerHistorian:QuestDb` (`Backend=QuestDb`) diff --git a/docs/Historian.md b/docs/Historian.md index 38d9bd37..643220ca 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -151,6 +151,9 @@ 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: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. | **`ServerHistorian:QuestDb`** (read only when `Backend=QuestDb`; bound by `HistorianBackendRegistry.BindQuestDb`): six keys — `ConnectionString` (required), `TablePrefix`, `RetentionDays`, `CommandTimeout`, `AutoCreateSchema`, `VisibilityTimeout`. The full table, defaults and validation @@ -772,6 +775,27 @@ The read client reports `GetHealthSnapshot()` (query/failure counters, consecuti `AutoCreateSchema=false`). A SQL error on a read surfaces as `BadHistoryOperationUnsupported` for **that node only**; a cancelled read (deadline, client abort) is not counted as a backend fault. +### Retention (`RetentionDays`) + +`hist.Sample` and `hist.Event` grow without bound unless `ServerHistorian:SqlServer:RetentionDays` is set. QuestDB already has an +equivalent TTL (`QuestDb:RetentionDays`); the Gateway backend's retention is the AVEVA Historian's own. + +- **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. +- **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 + until a batch comes back short. Cancellation is honoured between batches. The `hist.Tag` catalogue is never purged. +- **Shared databases.** Several pairs may point at one database; a session `sp_getapplock` (`OtOpcUa.Historian.Retention.`, + timeout 0) lets one purger run and the others skip that tick (Debug log). +- **Observability.** Counter `otopcua.historian.retention.purged{table=Sample|Event}`; a failed pass logs a Warning and retries next + interval — it never affects capture or reads. +- **Not done: partition switching.** For very large installs (billions of rows) a partitioned `hist.Sample` with sliding-window + `SWITCH`/drop is cheaper than row deletes; the purger is deliberately simple batched deletes. Another table (e.g. a future + `hist.SampleModified`, #668) is added by one more `DeleteLoopAsync` call in `SqlServerHistorianRetention.PurgeAsync`. + ### Docker-dev recipe The docker-dev `sql` container (`otopcua-dev-sql-1`, host port `14330`, `sa` with the committed dev password from @@ -795,7 +819,9 @@ sqlcmd -S localhost,14330 -U sa -P 'OtOpcUa!Dev123' -C -I -d OtOpcUaHistorian \ -Q "SELECT t.Name, COUNT(*) n, MAX(s.SourceTimeUtc) last FROM hist.Sample s JOIN hist.Tag t ON t.TagId=s.TagId GROUP BY t.Name" ``` -Only the Primary of a pair records, so the row count advances once per published value, not twice. Read it back with +Only the Primary of a pair records, so the row count advances once per published value, not twice. To watch retention, add +`ServerHistorian__SqlServer__RetentionDays: "1"` and `ServerHistorian__SqlServer__PurgeInterval: "00:00:30"`, insert a `hist.Sample` row +with an old `SourceTimeUtc` for an existing tag, and see it vanish within a tick while recent rows stay. Read it back with `Client.CLI historyread` and rewrite it with `Client.CLI historyupdate` ([Client.CLI.md](Client.CLI.md)). The password above is the committed docker-dev credential, not a secret; use `${secret:…}` / environment for anything real. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianOptions.cs index d5669b05..84a26732 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianOptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianOptions.cs @@ -22,6 +22,21 @@ public sealed partial class SqlServerHistorianOptions /// public bool AutoCreateSchema { get; set; } = true; + /// + /// Days of history to keep, measured on source time. null (default) keeps everything. When set, + /// a driver node deletes older hist.Sample/hist.Event rows every . + /// + public int? RetentionDays { get; set; } + + /// How often the retention purge runs. Must be positive. + public TimeSpan PurgeInterval { get; set; } = TimeSpan.FromHours(1); + + /// + /// Rows per DELETE TOP(n) batch. Default 4000 stays under SQL Server's 5000-lock escalation + /// threshold so a purge never escalates to a table lock. + /// + public int PurgeBatchSize { get; set; } = 4000; + [GeneratedRegex("^[A-Za-z_][A-Za-z0-9_]{0,127}$")] private static partial Regex IdentifierRegex(); @@ -41,6 +56,12 @@ public sealed partial class SqlServerHistorianOptions errors.Add("ServerHistorian:SqlServer:Schema must be a SQL identifier matching ^[A-Za-z_][A-Za-z0-9_]{0,127}$."); if (CommandTimeout <= TimeSpan.Zero) errors.Add("ServerHistorian:SqlServer:CommandTimeout must be greater than zero."); + if (RetentionDays is < 1) + errors.Add("ServerHistorian:SqlServer:RetentionDays must be at least 1 when set (omit it to keep history forever)."); + if (PurgeInterval <= TimeSpan.Zero) + errors.Add("ServerHistorian:SqlServer:PurgeInterval must be greater than zero."); + if (PurgeBatchSize < 1) + errors.Add("ServerHistorian:SqlServer:PurgeBatchSize must be at least 1."); return errors; } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs index 8f98031b..8076a9f3 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs @@ -96,5 +96,14 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider return new SqlServerHistorianDataUpdater(s.Conn, s.Catalog); } + /// + public IHistorianRetention? CreateRetention(IServiceProvider sp) + { + if (_options.RetentionDays is not { } days) return null; + var s = Get(sp); + return new SqlServerHistorianRetention(s.Conn, s.Bootstrap, _options.PurgeInterval, days, + _options.PurgeBatchSize, Loggers(sp).CreateLogger()); + } + private sealed record Shared(SqlHistorianConnection Conn, SqlServerSchemaBootstrapper Bootstrap, SqlServerTagCatalog Catalog); } 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 new file mode 100644 index 00000000..a729495e --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs @@ -0,0 +1,118 @@ +using System.Data; +using Microsoft.Data.SqlClient; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// +/// Deletes hist.Sample / hist.Event rows older than the retention window, in small +/// DELETE TOP(n) batches (under the lock-escalation threshold). Samples are purged per tag so each +/// delete is a seek on the clustered key (TagId, SourceTimeUtc); events seek on CX_Event_Time. +/// A session sp_getapplock (timeout 0) serialises purgers that share one database; a loser skips. +/// To purge another table later, add it to . +/// +internal sealed class SqlServerHistorianRetention : IHistorianRetention +{ + private readonly SqlHistorianConnection _conn; + private readonly SqlServerSchemaBootstrapper _bootstrap; + private readonly int _days; + private readonly int _batch; + private readonly TimeProvider _time; + private readonly ILogger _log; + + public SqlServerHistorianRetention(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, + TimeSpan interval, int retentionDays, int batchSize, ILogger? log = null, TimeProvider? time = null) + { + _conn = conn; + _bootstrap = bootstrap; + Interval = interval; + _days = retentionDays; + _batch = batchSize; + _log = log ?? NullLogger.Instance; + _time = time ?? TimeProvider.System; + } + + public TimeSpan Interval { get; } + + 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 + { + long samples = 0; + var tagIds = new List(); + await using (var cmd = _conn.CreateCommand(conn, $"SELECT TagId FROM {_conn.Q("Tag")}")) + await using (var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false)) + while (await r.ReadAsync(ct).ConfigureAwait(false)) tagIds.Add(r.GetInt32(0)); + + foreach (var id in tagIds) + { + samples += await DeleteLoopAsync(conn, + $"DELETE TOP (@n) FROM {_conn.Q("Sample")} WHERE TagId = @id AND SourceTimeUtc < @cutoff", + cutoff, id, ct).ConfigureAwait(false); + } + + var events = await DeleteLoopAsync(conn, + $"DELETE TOP (@n) FROM {_conn.Q("Event")} WHERE EventTimeUtc < @cutoff", + cutoff, null, ct).ConfigureAwait(false); + + return [new RetentionPurgeResult("Sample", samples), new RetentionPurgeResult("Event", events)]; + } + finally + { + await ReleaseAsync(conn).ConfigureAwait(false); + } + } + + private async Task DeleteLoopAsync(SqlConnection conn, string sql, DateTime cutoff, int? tagId, CancellationToken ct) + { + long total = 0; + while (true) + { + ct.ThrowIfCancellationRequested(); + await using var cmd = _conn.CreateCommand(conn, sql); + cmd.Parameters.Add(new SqlParameter("@n", SqlDbType.Int) { Value = _batch }); + if (tagId is { } id) cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id }); + SqlHistorianConnection.AddDateTime2(cmd, "@cutoff", cutoff); + var n = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + total += n; + if (n < _batch) return total; + } + } + + private string LockResource => $"OtOpcUa.Historian.Retention.{_conn.Schema}"; + + private async Task TryLockAsync(SqlConnection conn, CancellationToken ct) + { + await using var cmd = _conn.CreateCommand(conn, + "DECLARE @r int; EXEC @r = sp_getapplock @Resource = @res, @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT @r;"); + cmd.Parameters.Add(new SqlParameter("@res", SqlDbType.NVarChar, 255) { Value = LockResource }); + return Convert.ToInt32(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false)) >= 0; + } + + private async Task ReleaseAsync(SqlConnection conn) + { + try + { + await using var cmd = _conn.CreateCommand(conn, + "EXEC sp_releaseapplock @Resource = @res, @LockOwner = 'Session';"); + cmd.Parameters.Add(new SqlParameter("@res", SqlDbType.NVarChar, 255) { Value = LockResource }); + await cmd.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + _log.LogDebug(ex, "Releasing the retention applock failed."); + } + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 416052f2..29dd14a3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -326,6 +326,14 @@ if (hasDriver) builder.Services.AddSingleton(sp => sp.GetRequiredService().CreateValueWriter(sp)); } + // Historian retention (#667). Only a backend that manages retention in-process returns a purger + // (SqlServer with RetentionDays set); otherwise the service is inert. Primary-gated via IHistorizationRoleView. + if (serverHistorianOptions.Enabled) + builder.Services.AddHostedService(sp => new HistorianRetentionService( + sp.GetRequiredService().CreateRetention(sp), + sp.GetService(), + sp.GetService>())); + // Bind every cross-platform driver factory before AddAkka resolves IDriverFactory — replaces // the F7-default NullDriverFactory with a real DriverFactoryRegistryAdapter so DriverHostActor // can materialise real IDriver instances on deploy. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianRetentionService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianRetentionService.cs new file mode 100644 index 00000000..52f3edf1 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianRetentionService.cs @@ -0,0 +1,95 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; +using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian; + +/// +/// Timer that drives an purge. Each tick is gated on +/// so only the node that is writing +/// history purges it (an unknown role allows, matching the capture gate). A null retention (backend has +/// none, or no window configured) makes the service inert. Never throws out of the loop. +/// +public sealed class HistorianRetentionService : BackgroundService +{ + /// Counter of rows purged, tagged table. + public const string PurgedCounterName = "otopcua.historian.retention.purged"; + + private static readonly Counter Purged = OtOpcUaTelemetry.Meter.CreateCounter( + PurgedCounterName, unit: "{row}", description: "Historian rows deleted by the retention purge."); + + private readonly IHistorianRetention? _retention; + private readonly IHistorizationRoleView? _roleView; + private readonly ILogger _log; + private readonly KeyValuePair[] _extraTags; + + /// Creates the service. + /// Backend purge seam; null makes the service inert. + /// Primary gate; null means always run. + /// Logger. + /// Extra tags added to every counter increment (tests use a unique one). + public HistorianRetentionService(IHistorianRetention? retention, IHistorizationRoleView? roleView, + ILogger? log = null, params KeyValuePair[] extraTags) + { + _retention = retention; + _roleView = roleView; + _log = (ILogger?)log ?? NullLogger.Instance; + _extraTags = extraTags; + } + + /// Runs one gated tick. Never throws. + /// Cancellation. + /// True when a purge pass ran. + public async Task TickAsync(CancellationToken ct) + { + if (_retention is null) return false; + if (_roleView is { ShouldRecordHistorizedValues: false }) + { + _log.LogDebug("Historian retention purge skipped: this node is not the recording node."); + return false; + } + + try + { + var results = await _retention.PurgeAsync(ct).ConfigureAwait(false); + foreach (var r in results) + { + if (r.Rows <= 0) continue; + var tags = new TagList { { "table", r.Table } }; + foreach (var t in _extraTags) tags.Add(t); + Purged.Add(r.Rows, tags); + _log.LogInformation("Historian retention purged {Rows} rows from {Table}.", r.Rows, r.Table); + } + return true; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return false; + } + catch (Exception ex) + { + _log.LogWarning(ex, "Historian retention purge failed; will retry next interval."); + return false; + } + } + + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + if (_retention is null) return; + try + { + using var timer = new PeriodicTimer(_retention.Interval); + await TickAsync(stoppingToken).ConfigureAwait(false); + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)) + await TickAsync(stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + } + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianBackendProvider.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianBackendProvider.cs index 3762ff77..c2c501b2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianBackendProvider.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianBackendProvider.cs @@ -39,4 +39,9 @@ public interface IHistorianBackendProvider /// Service provider. /// The updater, or null when the backend does not support HistoryUpdate. IHistorianDataUpdater? CreateUpdater(IServiceProvider sp); + + /// Creates the retention purge seam, if the backend manages retention in-process. + /// Service provider. + /// The purger, or null (default) when the backend has no retention or enforces it itself. + IHistorianRetention? CreateRetention(IServiceProvider sp) => null; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianRetention.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianRetention.cs new file mode 100644 index 00000000..c6138da5 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianRetention.cs @@ -0,0 +1,25 @@ +namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian; + +/// Rows removed from one historian table by a single purge pass. +/// Unqualified table name (the table tag on the retention counter). +/// Rows deleted. +public sealed record RetentionPurgeResult(string Table, long Rows); + +/// +/// Backend-side retention purge: deletes history older than the backend's configured retention window. +/// Created by and driven on a timer by +/// , which owns scheduling and the Primary gate. +/// +public interface IHistorianRetention +{ + /// How often should run a pass. + TimeSpan Interval { get; } + + /// + /// Runs one purge pass. Returns an empty list when the pass was skipped (another process holds the + /// purge lock). Implementations honour cancellation between batches. + /// + /// Cancellation. + /// Per-table deleted-row counts. + Task> PurgeAsync(CancellationToken ct); +} 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 new file mode 100644 index 00000000..4dc56b97 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs @@ -0,0 +1,133 @@ +using Microsoft.Data.SqlClient; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +[Collection(SqlHistorianCollection.Name)] +public sealed class RetentionTests(SqlHistorianFixture fx) +{ + private static readonly DateTime Now = new(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc); + + private sealed class FixedTime(DateTime utc) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => new(utc, TimeSpan.Zero); + } + + private (SqlHistorianConnection C, SqlServerSchemaBootstrapper B, SqlServerHistorianOptions O, string S) Make() + { + var s = SqlHistorianFixture.UniqueSchema(); + var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s }; + var c = new SqlHistorianConnection(o); + return (c, new SqlServerSchemaBootstrapper(c, o), o, s); + } + + private async Task ExecAsync(string sql) + { + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = sql; + await cmd.ExecuteNonQueryAsync(); + } + + private async Task CountAsync(string sql) + { + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = sql; + return Convert.ToInt64(await cmd.ExecuteScalarAsync()); + } + + /// 3 tags x (5 old + 2 new) samples, plus 4 old + 1 new events. Retention 30 days at . + private async Task SeedAsync(SqlServerSchemaBootstrapper b, string s) + { + await b.EnsureAsync(CancellationToken.None); + var old = Now.AddDays(-40).ToString("yyyy-MM-ddTHH:mm:ss"); + var fresh = Now.AddDays(-1).ToString("yyyy-MM-ddTHH:mm:ss"); + var sql = new System.Text.StringBuilder(); + for (var t = 1; t <= 3; t++) + { + sql.Append($"INSERT [{s}].Tag (Name, DataType, CreatedUtc, UpdatedUtc) VALUES (N'tag{t}', 1, SYSUTCDATETIME(), SYSUTCDATETIME());"); + } + sql.Append($"DECLARE @i int = 0; WHILE @i < 5 BEGIN INSERT [{s}].Sample (TagId, SourceTimeUtc, ValueKind, NumValue, StatusCode, ServerTimeUtc) SELECT TagId, DATEADD(minute, @i, '{old}'), 1, 1.0, 0, SYSUTCDATETIME() FROM [{s}].Tag; SET @i += 1; END;"); + sql.Append($"SET @i = 0; WHILE @i < 2 BEGIN INSERT [{s}].Sample (TagId, SourceTimeUtc, ValueKind, NumValue, StatusCode, ServerTimeUtc) SELECT TagId, DATEADD(minute, @i, '{fresh}'), 1, 1.0, 0, SYSUTCDATETIME() FROM [{s}].Tag; SET @i += 1; END;"); + sql.Append($"SET @i = 0; WHILE @i < 4 BEGIN INSERT [{s}].Event (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, EventTimeUtc, ReceivedTimeUtc) VALUES (CONCAT('old', @i), 's', 'a', 'n', 't', 1, 'Activated', DATEADD(minute, @i, '{old}'), SYSUTCDATETIME()); SET @i += 1; END;"); + sql.Append($"INSERT [{s}].Event (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, EventTimeUtc, ReceivedTimeUtc) VALUES ('new0', 's', 'a', 'n', 't', 1, 'Activated', '{fresh}', SYSUTCDATETIME());"); + await ExecAsync(sql.ToString()); + } + + [Theory] + [InlineData(4000)] + [InlineData(1)] // forces the delete loop to iterate once per row + public async Task Purge_removes_only_rows_older_than_the_window(int batch) + { + fx.SkipIfUnavailable(); + var (c, b, _, s) = Make(); + await SeedAsync(b, s); + + var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, batch, time: new FixedTime(Now)); + var results = await purger.PurgeAsync(CancellationToken.None); + + results.ShouldBe([new RetentionPurgeResult("Sample", 15), new RetentionPurgeResult("Event", 4)]); + (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(6); + (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample WHERE SourceTimeUtc < '{Now.AddDays(-30):yyyy-MM-ddTHH:mm:ss}'")).ShouldBe(0); + (await CountAsync($"SELECT COUNT(*) FROM [{s}].Event")).ShouldBe(1); + (await CountAsync($"SELECT COUNT(*) FROM [{s}].Tag")).ShouldBe(3); + } + + [Fact] + public async Task A_second_purge_is_a_no_op() + { + 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); + var again = await purger.PurgeAsync(CancellationToken.None); + again.ShouldBe([new RetentionPurgeResult("Sample", 0), new RetentionPurgeResult("Event", 0)]); + } + + [Fact] + public async Task A_purger_skips_while_another_holds_the_applock() + { + fx.SkipIfUnavailable(); + var (c, b, _, s) = Make(); + await SeedAsync(b, s); + + // Hold the same session applock the purger takes, on a separate connection. + await using var holder = new SqlConnection(fx.ConnectionString); + await holder.OpenAsync(); + await using (var cmd = holder.CreateCommand()) + { + cmd.CommandText = $"EXEC sp_getapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0;"; + await cmd.ExecuteNonQueryAsync(); + } + + var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now)); + (await purger.PurgeAsync(CancellationToken.None)).ShouldBeEmpty(); + (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(21); + + await using (var cmd = holder.CreateCommand()) + { + cmd.CommandText = $"EXEC sp_releaseapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockOwner = 'Session';"; + await cmd.ExecuteNonQueryAsync(); + } + (await purger.PurgeAsync(CancellationToken.None)).ShouldNotBeEmpty(); + (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(6); + } + + [Fact] + public async Task Cancellation_stops_the_purge() + { + 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)); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianRetentionOptionsTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianRetentionOptionsTests.cs new file mode 100644 index 00000000..a04614ce --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianRetentionOptionsTests.cs @@ -0,0 +1,62 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests; + +public sealed class SqlServerHistorianRetentionOptionsTests +{ + private static SqlServerHistorianOptions Valid() => new() { ConnectionString = "Server=x" }; + + [Fact] + public void Retention_is_off_by_default_with_sane_defaults() + { + var o = Valid(); + o.RetentionDays.ShouldBeNull(); + o.PurgeInterval.ShouldBe(TimeSpan.FromHours(1)); + o.PurgeBatchSize.ShouldBe(4000); + o.Validate().ShouldBeEmpty(); + } + + [Theory] + [InlineData(0)] + [InlineData(-5)] + public void Non_positive_RetentionDays_is_rejected(int days) + { + var o = Valid(); + o.RetentionDays = days; + o.Validate().ShouldContain(e => e.Contains("RetentionDays")); + } + + [Fact] + public void Positive_RetentionDays_is_valid() + { + var o = Valid(); + o.RetentionDays = 30; + o.Validate().ShouldBeEmpty(); + } + + [Fact] + public void Bad_interval_and_batch_are_rejected() + { + var o = Valid(); + o.PurgeInterval = TimeSpan.Zero; + o.PurgeBatchSize = 0; + var errors = o.Validate(); + errors.ShouldContain(e => e.Contains("PurgeInterval")); + errors.ShouldContain(e => e.Contains("PurgeBatchSize")); + } + + [Fact] + public void Provider_creates_retention_only_when_RetentionDays_set() + { + var o = Valid(); + new SqlServerHistorianProvider(o).CreateRetention(new EmptyServices()).ShouldBeNull(); + o.RetentionDays = 7; + new SqlServerHistorianProvider(o).CreateRetention(new EmptyServices()).ShouldNotBeNull(); + } + + private sealed class EmptyServices : IServiceProvider + { + public object? GetService(Type serviceType) => null; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianRetentionServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianRetentionServiceTests.cs new file mode 100644 index 00000000..25d0384a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/HistorianRetentionServiceTests.cs @@ -0,0 +1,95 @@ +using System.Diagnostics.Metrics; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; +using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian; + +public sealed class HistorianRetentionServiceTests +{ + private sealed class FakeRetention(Func>> run) : IHistorianRetention + { + public int Calls; + public TimeSpan Interval => TimeSpan.FromHours(1); + public Task> PurgeAsync(CancellationToken ct) + { + Interlocked.Increment(ref Calls); + return run(); + } + } + + private sealed class FakeRole(bool record) : IHistorizationRoleView + { + public bool ShouldRecordHistorizedValues => record; + public void PublishHistorizationRole(bool shouldRecordHistorizedValues) { } + } + + private static Task> Rows(long sample, long ev) => + Task.FromResult>( + [new RetentionPurgeResult("Sample", sample), new RetentionPurgeResult("Event", ev)]); + + [Fact] + public async Task Runs_when_role_is_unknown_and_when_recording() + { + var r1 = new FakeRetention(() => Rows(0, 0)); + (await new HistorianRetentionService(r1, null).TickAsync(CancellationToken.None)).ShouldBeTrue(); + var r2 = new FakeRetention(() => Rows(0, 0)); + (await new HistorianRetentionService(r2, new FakeRole(true)).TickAsync(CancellationToken.None)).ShouldBeTrue(); + r1.Calls.ShouldBe(1); + r2.Calls.ShouldBe(1); + } + + [Fact] + public async Task Skips_when_not_the_recording_node() + { + var r = new FakeRetention(() => Rows(1, 1)); + (await new HistorianRetentionService(r, new FakeRole(false)).TickAsync(CancellationToken.None)).ShouldBeFalse(); + r.Calls.ShouldBe(0); + } + + [Fact] + public async Task Null_retention_is_inert() + { + (await new HistorianRetentionService(null, null).TickAsync(CancellationToken.None)).ShouldBeFalse(); + } + + [Fact] + public async Task A_throwing_purge_does_not_escape() + { + var r = new FakeRetention(() => throw new InvalidOperationException("boom")); + (await new HistorianRetentionService(r, null).TickAsync(CancellationToken.None)).ShouldBeFalse(); + r.Calls.ShouldBe(1); + } + + [Fact] + public async Task Emits_purged_counter_per_table_and_skips_zero_rows() + { + var marker = Guid.NewGuid().ToString("N"); + var seen = new List<(string Table, long Value)>(); + using var listener = new MeterListener(); + listener.InstrumentPublished = (inst, l) => + { + if (inst.Meter.Name == OtOpcUaTelemetry.MeterName && inst.Name == HistorianRetentionService.PurgedCounterName) + l.EnableMeasurementEvents(inst); + }; + listener.SetMeasurementEventCallback((_, value, tags, _) => + { + string? table = null; var mine = false; + foreach (var t in tags) + { + if (t.Key == "table") table = (string?)t.Value; + if (t.Key == "test" && (string?)t.Value == marker) mine = true; + } + if (mine) lock (seen) seen.Add((table!, value)); + }); + listener.Start(); + + var r = new FakeRetention(() => Rows(7, 0)); + var svc = new HistorianRetentionService(r, null, null, new KeyValuePair("test", marker)); + await svc.TickAsync(CancellationToken.None); + + lock (seen) seen.ShouldBe([("Sample", 7L)]); + } +} From e4123ada91bc8c8e73d162ce3c959cb0e5c43105 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 08:27:31 -0400 Subject: [PATCH 2/2] fix(historian): retention applock release hardening, shutdown-quiet cancellation, release tests, docs (#667 review) --- docs/Historian.md | 9 ++-- .../SqlServerHistorianRetention.cs | 23 ++++++--- .../Historian/HistorianRetentionService.cs | 3 +- .../RetentionTests.cs | 49 +++++++++++++++++-- 4 files changed, 69 insertions(+), 15 deletions(-) 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(); } }