From 9ae7b9c781e4837afcef7676edcb1daf6f0b8ef7 Mon Sep 17 00:00:00 2001 From: John Simons Date: Fri, 28 Aug 2026 15:49:41 +1000 Subject: [PATCH 1/7] Add retention sweep metrics for cycle duration, rows deleted, and consecutive failures Persisters publishing onto a host's already-registered meter cannot reference the host assembly, so meter names are extracted into a shared ServiceControlMeters constants class. Cycles interrupted by cancellation are excluded from measurements, as a clean shutdown is not a failure worth alerting on. --- .../Auditing/Metrics/IngestionMetrics.cs | 3 +- .../ServiceControlMeters.cs | 12 ++ .../Abstractions/BasePersistence.cs | 1 + .../Infrastructure/RetentionMetrics.cs | 120 +++++++++++ .../Infrastructure/RetentionSweeper.cs | 22 +- .../EFCore/RecordedRetentionMetrics.cs | 91 +++++++++ .../EFCore/RetentionMetricsTests.cs | 191 ++++++++++++++++++ .../EFCore/RetentionSweepTests.cs | 45 +++++ .../Operations/Metrics/IngestionMetrics.cs | 3 +- 9 files changed, 484 insertions(+), 4 deletions(-) create mode 100644 src/ServiceControl.Infrastructure/ServiceControlMeters.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs diff --git a/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs b/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs index d79a7e76f0..038fefb7df 100644 --- a/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs +++ b/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs @@ -7,11 +7,12 @@ namespace ServiceControl.Audit.Auditing.Metrics; using EndpointPlugin.Messages.SagaState; using NServiceBus; using NServiceBus.Transport; +using ServiceControl.Infrastructure; using ServiceControl.Infrastructure.Ingestion.Metrics; public class IngestionMetrics { - public const string MeterName = "Particular.ServiceControl.Audit"; + public const string MeterName = ServiceControlMeters.Audit; public static readonly string BatchDurationInstrumentName = $"{InstrumentPrefix}.batch_duration_seconds"; public static readonly string MessageDurationInstrumentName = $"{InstrumentPrefix}.message_duration_seconds"; diff --git a/src/ServiceControl.Infrastructure/ServiceControlMeters.cs b/src/ServiceControl.Infrastructure/ServiceControlMeters.cs new file mode 100644 index 0000000000..aaac716f21 --- /dev/null +++ b/src/ServiceControl.Infrastructure/ServiceControlMeters.cs @@ -0,0 +1,12 @@ +namespace ServiceControl.Infrastructure; + +/// +/// The meters each instance publishes on. Shared because persisters publish onto the meter their +/// host has already registered with the exporter, and the two assemblies cannot reference each +/// other. +/// +public static class ServiceControlMeters +{ + public const string Error = "Particular.ServiceControl"; + public const string Audit = "Particular.ServiceControl.Audit"; +} diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index a1ced0ed8e..4a0195210d 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -37,6 +37,7 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste if (settings.RunRetentionSweep) { + services.AddSingleton(); services.AddHostedService(); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs new file mode 100644 index 0000000000..34e384de7b --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs @@ -0,0 +1,120 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using System.Diagnostics; +using System.Diagnostics.Metrics; +using ServiceControl.Infrastructure; + +public enum RetentionEntity +{ + FailedMessages, + EventLog, + GroupComments +} + +public class RetentionMetrics +{ + public const string MeterName = ServiceControlMeters.Error; + + public static readonly string CycleDurationInstrumentName = $"{InstrumentPrefix}.cycle_duration_seconds"; + public static readonly string RowsDeletedInstrumentName = $"{InstrumentPrefix}.rows_deleted_total"; + public static readonly string ConsecutiveFailuresInstrumentName = $"{InstrumentPrefix}.consecutive_failures_total"; + + public RetentionMetrics(IMeterFactory meterFactory) + { + var meter = meterFactory.Create(MeterName, MeterVersion); + + cycleDuration = meter.CreateHistogram( + CycleDurationInstrumentName, + unit: "seconds", + description: "Retention sweep pass duration in seconds", + tags: null, + // A sweep pass is sub-second when it is keeping up and minutes long when it is working + // through a backlog, so the default boundaries resolve neither end. + advice: new InstrumentAdvice { HistogramBucketBoundaries = [0.1, 0.5, 1, 5, 15, 60, 300, 900] }); + + rowsDeleted = meter.CreateCounter(RowsDeletedInstrumentName, description: "Rows deleted by the retention sweep"); + consecutiveFailureGauge = meter.CreateObservableGauge(ConsecutiveFailuresInstrumentName, ObserveConsecutiveFailures, description: "Consecutive retention sweep failures"); + } + + public RetentionCycleMetrics BeginCycle(RetentionEntity entity, CancellationToken cancellationToken = default) => new(this, entity, cancellationToken); + + public void RecordRowsDeleted(RetentionEntity entity, int rows) => rowsDeleted.Add(rows, EntityTags[(int)entity]); + + internal void RecordCycle(RetentionEntity entity, TimeSpan elapsed, bool success) + { + var tags = EntityTags[(int)entity]; + tags.Add("result", success ? "success" : "failed"); + + cycleDuration.Record(elapsed.TotalSeconds, tags); + + if (success) + { + Interlocked.Exchange(ref consecutiveFailures[(int)entity], 0); + } + else + { + Interlocked.Increment(ref consecutiveFailures[(int)entity]); + } + } + + IEnumerable> ObserveConsecutiveFailures() + { + for (var entity = 0; entity < consecutiveFailures.Length; entity++) + { + yield return new Measurement(Volatile.Read(ref consecutiveFailures[entity]), EntityTags[entity]); + } + } + + static TagList EntityTag(string entity) => new() { { "retention.entity", entity } }; + + readonly long[] consecutiveFailures = new long[EntityTags.Length]; + + readonly Histogram cycleDuration; + readonly Counter rowsDeleted; +#pragma warning disable IDE0052 + readonly ObservableGauge consecutiveFailureGauge; +#pragma warning restore IDE0052 + + static readonly TagList[] EntityTags = + [ + EntityTag("failed_messages"), + EntityTag("event_log"), + EntityTag("group_comments") + ]; + + const string MeterVersion = "0.1.0"; + const string InstrumentPrefix = "sc.retention"; +} + +/// +/// One pass of the retention sweep. A pass interrupted by shutdown is not a measurement of +/// anything, so a cancelled cycle records neither a duration nor a failure. +/// +public sealed class RetentionCycleMetrics : IDisposable +{ + internal RetentionCycleMetrics(RetentionMetrics metrics, RetentionEntity entity, CancellationToken cancellationToken) + { + this.metrics = metrics; + this.entity = entity; + this.cancellationToken = cancellationToken; + } + + public void Complete() => completed = true; + + public void Dispose() + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + metrics.RecordCycle(entity, stopwatch.Elapsed, completed); + } + + bool completed; + + readonly RetentionMetrics metrics; + readonly RetentionEntity entity; + readonly CancellationToken cancellationToken; + readonly Stopwatch stopwatch = Stopwatch.StartNew(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs index 07811c4ad3..fd57ae2faa 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs @@ -17,6 +17,7 @@ public class RetentionSweeper( TimeProvider timeProvider, IServiceScopeFactory serviceScopeFactory, IBodyStoragePersistence bodyStorage, + RetentionMetrics metrics, EFPersisterSettings settings) : BackgroundService { const int BatchSize = 1000; @@ -71,18 +72,25 @@ async Task Sweep(bool pace, CancellationToken cancellationToken) // group ids are deterministic, reattach a stale comment if the same failure ever recurs. async Task SweepOrphanedGroupComments(CancellationToken cancellationToken) { + using var cycle = metrics.BeginCycle(RetentionEntity.GroupComments, cancellationToken); using var scope = serviceScopeFactory.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); - await dbContext.GroupComments + var deleted = await dbContext.GroupComments .Where(comment => !dbContext.FailedMessageGroups.Any(group => group.GroupId == comment.GroupId)) .ExecuteDeleteAsync(cancellationToken); + + metrics.RecordRowsDeleted(RetentionEntity.GroupComments, deleted); + + cycle.Complete(); } // Event log items are insert-only and carry no external bodies, so each batch is a single // ordered DELETE. async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) { + using var cycle = metrics.BeginCycle(RetentionEntity.EventLog, cancellationToken); + var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.EventsRetentionPeriod; while (!cancellationToken.IsCancellationRequested) @@ -96,6 +104,8 @@ async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) .Take(BatchSize) .ExecuteDeleteAsync(cancellationToken); + metrics.RecordRowsDeleted(RetentionEntity.EventLog, deleted); + if (deleted < BatchSize) { break; @@ -106,10 +116,14 @@ async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) await Task.Delay(BatchPause, timeProvider, cancellationToken); } } + + cycle.Complete(); } async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) { + using var cycle = metrics.BeginCycle(RetentionEntity.FailedMessages, cancellationToken); + var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.ErrorRetentionPeriod; while (!cancellationToken.IsCancellationRequested) @@ -142,11 +156,13 @@ async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) // The predicate is re-asserted so a message that was re-failed (back to Unresolved) // between the select and the delete is left alone. The cascade removes its group rows. - await dbContext.FailedMessages + var deleted = await dbContext.FailedMessages .Where(failedMessage => ids.Contains(failedMessage.UniqueMessageId)) .Where(IsExpired(cutoff)) .ExecuteDeleteAsync(cancellationToken); + metrics.RecordRowsDeleted(RetentionEntity.FailedMessages, deleted); + if (expired.Count < BatchSize) { break; @@ -157,6 +173,8 @@ await dbContext.FailedMessages await Task.Delay(BatchPause, timeProvider, cancellationToken); } } + + cycle.Complete(); } async Task DeleteExternalBody(Guid uniqueMessageId, CancellationToken cancellationToken) diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs b/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs new file mode 100644 index 0000000000..687164d117 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs @@ -0,0 +1,91 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Diagnostics.Metrics; +using System.Linq; +using ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// Collects everything the retention instruments record, for the meter belonging to one factory. +/// Every fixture in the run shares the meter name, so the factory is what tells these instruments +/// apart from the ones another test left behind. +/// +sealed class RecordedRetentionMetrics : IDisposable +{ + public RecordedRetentionMetrics(IMeterFactory meterFactory) + { + listener = new MeterListener + { + InstrumentPublished = (instrument, activeListener) => + { + if (instrument.Meter.Name == RetentionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, meterFactory)) + { + activeListener.EnableMeasurementEvents(instrument); + } + } + }; + + listener.SetMeasurementEventCallback((instrument, measurement, tags, _) => Add(instrument, measurement, tags)); + listener.SetMeasurementEventCallback((instrument, measurement, tags, _) => Add(instrument, measurement, tags)); + listener.Start(); + } + + public IReadOnlyList Of(string instrumentName, RetentionEntity entity) + { + lock (measurements) + { + return + [ + .. measurements.Where(measurement => + measurement.InstrumentName == instrumentName && + Equals(measurement.Tags["retention.entity"], EntityTag(entity))) + ]; + } + } + + public IReadOnlyList Cycles(RetentionEntity entity) => Of(RetentionMetrics.CycleDurationInstrumentName, entity); + + public double RowsDeleted(RetentionEntity entity) => + Of(RetentionMetrics.RowsDeletedInstrumentName, entity).Sum(measurement => measurement.Value); + + public double ConsecutiveFailures(RetentionEntity entity) + { + listener.RecordObservableInstruments(); + + return Of(RetentionMetrics.ConsecutiveFailuresInstrumentName, entity)[^1].Value; + } + + public void Dispose() => listener.Dispose(); + + void Add(Instrument instrument, double value, ReadOnlySpan> tags) + { + var copied = new Dictionary(); + + foreach (var tag in tags) + { + copied[tag.Key] = tag.Value; + } + + lock (measurements) + { + measurements.Add(new Recorded(instrument.Name, value, copied)); + } + } + + static string EntityTag(RetentionEntity entity) => entity switch + { + RetentionEntity.FailedMessages => "failed_messages", + RetentionEntity.EventLog => "event_log", + RetentionEntity.GroupComments => "group_comments", + _ => throw new ArgumentOutOfRangeException(nameof(entity)) + }; + + readonly MeterListener listener; + readonly List measurements = []; + + public sealed record Recorded(string InstrumentName, double Value, Dictionary Tags) + { + public object Result => Tags["result"]; + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs new file mode 100644 index 0000000000..774f5a8a0a --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs @@ -0,0 +1,191 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Diagnostics.Metrics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// Instrument names are what dashboards and alerts are built on, so they are a published contract +/// and not an implementation detail. +/// +[TestFixture] +class RetentionMetricsTests +{ + [SetUp] + public void CreateMeterFactory() => provider = new ServiceCollection().AddMetrics().BuildServiceProvider(); + + [TearDown] + public void DisposeMeterFactory() => provider.Dispose(); + + [Test] + public void The_meter_publishes_the_instruments_it_is_named_for() + { + var published = new List(); + + using var listener = new MeterListener + { + InstrumentPublished = (instrument, _) => + { + if (BelongsToThisTest(instrument)) + { + published.Add(instrument.Name); + } + } + }; + + listener.Start(); + + _ = new RetentionMetrics(MeterFactory); + + Assert.That(published.Order(), Is.EqualTo(new[] + { + "sc.retention.consecutive_failures_total", + "sc.retention.cycle_duration_seconds", + "sc.retention.rows_deleted_total" + })); + } + + [Test] + public void A_completed_cycle_is_recorded_as_a_success() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + using (var cycle = metrics.BeginCycle(RetentionEntity.EventLog)) + { + cycle.Complete(); + } + + var cycles = recorded.Cycles(RetentionEntity.EventLog); + + using (Assert.EnterMultipleScope()) + { + Assert.That(cycles, Has.Count.EqualTo(1)); + Assert.That(cycles[0].Result, Is.EqualTo("success")); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.EventLog), Is.Zero); + } + } + + [Test] + public void An_abandoned_cycle_is_recorded_as_a_failure() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(RetentionEntity.EventLog).Dispose(); + + var cycles = recorded.Cycles(RetentionEntity.EventLog); + + using (Assert.EnterMultipleScope()) + { + Assert.That(cycles, Has.Count.EqualTo(1)); + Assert.That(cycles[0].Result, Is.EqualTo("failed")); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.EventLog), Is.EqualTo(1)); + } + } + + [Test] + public void Consecutive_failures_are_counted_per_entity() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose(); + metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose(); + metrics.BeginCycle(RetentionEntity.EventLog).Dispose(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.EqualTo(2)); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.EventLog), Is.EqualTo(1)); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.GroupComments), Is.Zero); + } + } + + [Test] + public void A_success_clears_the_failures_of_that_entity_alone() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose(); + metrics.BeginCycle(RetentionEntity.EventLog).Dispose(); + + using (var cycle = metrics.BeginCycle(RetentionEntity.FailedMessages)) + { + cycle.Complete(); + } + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.Zero); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.EventLog), Is.EqualTo(1)); + } + } + + [Test] + public void A_cycle_interrupted_by_shutdown_is_not_recorded() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + using var shutdown = new CancellationTokenSource(); + + using (metrics.BeginCycle(RetentionEntity.FailedMessages, shutdown.Token)) + { + shutdown.Cancel(); + } + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.Cycles(RetentionEntity.FailedMessages), Is.Empty); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.Zero); + } + } + + [Test] + public void Deleted_rows_are_counted_per_entity() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.RecordRowsDeleted(RetentionEntity.FailedMessages, 1000); + metrics.RecordRowsDeleted(RetentionEntity.FailedMessages, 7); + metrics.RecordRowsDeleted(RetentionEntity.GroupComments, 3); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.RowsDeleted(RetentionEntity.FailedMessages), Is.EqualTo(1007)); + Assert.That(recorded.RowsDeleted(RetentionEntity.GroupComments), Is.EqualTo(3)); + } + } + + [Test] + public void Concurrent_failures_are_all_counted() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + const int failedCycles = 1000; + + Parallel.For(0, failedCycles, _ => metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose()); + + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.EqualTo(failedCycles)); + } + + RecordedRetentionMetrics Listen() => new(MeterFactory); + + // Every fixture in the run shares the meter name, so the factory is what tells the instruments + // created here apart from the ones another test left behind. + bool BelongsToThisTest(Instrument instrument) => + instrument.Meter.Name == RetentionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, MeterFactory); + + IMeterFactory MeterFactory => provider.GetRequiredService(); + + ServiceProvider provider; +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs index 122677d200..8948999fe6 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs @@ -2,12 +2,15 @@ namespace ServiceControl.Persistence.Tests; using System; using System.Collections.Generic; +using System.Diagnostics.Metrics; using System.Linq; using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; using NUnit.Framework; using ServiceControl.EventLog; using ServiceControl.MessageFailures; using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.Infrastructure; class RetentionSweepTests : ErrorIngestionTestBase @@ -168,6 +171,48 @@ public async Task Archived_messages_are_swept_after_the_archiver_updates_the_tim Assert.That(await FindFailedMessage(messageId), Is.Null); } + [Test] + public async Task Counts_the_rows_it_deletes() + { + EFSettings.EventsRetentionPeriod = TimeSpan.FromDays(14); + + await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-31)); + await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-29)); + await Store(EventLogRow("expired", Now.AddDays(-15))); + + var expiredWithGroup = await SeedFailedMessage(FailedMessageStatus.Archived, Now.AddDays(-31)); + await GroupsStore.EditComment(await SeedGroup(expiredWithGroup), "Raised with the shipping team"); + + using var recorded = ListenToRetentionMetrics(); + + await RunRetentionSweep(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.RowsDeleted(RetentionEntity.FailedMessages), Is.EqualTo(2), "the 29 day old message is still within retention"); + Assert.That(recorded.RowsDeleted(RetentionEntity.EventLog), Is.EqualTo(1)); + Assert.That(recorded.RowsDeleted(RetentionEntity.GroupComments), Is.EqualTo(1)); + } + } + + [Test] + public async Task Records_a_successful_cycle_for_every_pass() + { + using var recorded = ListenToRetentionMetrics(); + + await RunRetentionSweep(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.Cycles(RetentionEntity.FailedMessages).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.Cycles(RetentionEntity.EventLog).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.Cycles(RetentionEntity.GroupComments).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.Zero); + } + } + + RecordedRetentionMetrics ListenToRetentionMetrics() => new(ServiceProvider.GetRequiredService()); + async Task SeedGroup(Guid uniqueMessageId) { var groupId = Guid.NewGuid().ToString(); diff --git a/src/ServiceControl/Operations/Metrics/IngestionMetrics.cs b/src/ServiceControl/Operations/Metrics/IngestionMetrics.cs index b877a21164..e395919797 100644 --- a/src/ServiceControl/Operations/Metrics/IngestionMetrics.cs +++ b/src/ServiceControl/Operations/Metrics/IngestionMetrics.cs @@ -5,11 +5,12 @@ namespace ServiceControl.Operations.Metrics; using System.Diagnostics.Metrics; using System.Threading; using NServiceBus.Transport; +using ServiceControl.Infrastructure; using ServiceControl.Infrastructure.Ingestion.Metrics; public class IngestionMetrics { - public const string MeterName = "Particular.ServiceControl"; + public const string MeterName = ServiceControlMeters.Error; public static readonly string BatchDurationInstrumentName = $"{InstrumentPrefix}.batch_duration_seconds"; public static readonly string MessageDurationInstrumentName = $"{InstrumentPrefix}.message_duration_seconds"; From 9c1fe14744e70176d5e8f384139a04551b41497a Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 31 Aug 2026 15:20:25 +1000 Subject: [PATCH 2/7] Document retention sweep metrics for SQL persisters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Covers the three metrics (cycle duration, rows deleted, consecutive failures), their tags, example PromQL queries, and guidance on reading the shapes — including what a climbing consecutive-failures gauge means for body store permissions and why passes are isolated from one another. --- docs/telemetry.md | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/docs/telemetry.md b/docs/telemetry.md index 94b3c1ca10..ff2bc9939d 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -66,6 +66,36 @@ What the shapes mean when tuning: Example Grafana dashboard - https://github.com/andreasohlund/Docker/blob/main/otel-monitoring/grafana-platform-template.json +## Retention + +Only the SQL persisters report retention. On RavenDB the expiry is metadata on each document and the deletes happen inside the Raven server, so there is no sweep of ours to measure. + +Meter `Particular.ServiceControl`, the same meter the instance uses for its ingestion. The prefix carries no instance segment, so when the audit SQL persister reports its own retention it uses these same names, and `job` (or `exported_job`) separates the two. + +The sweep runs hourly and makes one pass per kind of row. Each pass is measured on its own, tagged `retention.entity`: `failed_messages`, `event_log` or `group_comments`. + +- `sc.retention.cycle_duration_seconds` - Retention sweep pass duration in seconds + - `retention.entity` - Which pass this was + - `result` - The outcome: `success`, `failed`, or `cancelled` if shutdown cut the pass short +- `sc.retention.rows_deleted_total` - Rows deleted by the retention sweep + - `retention.entity` - Which pass deleted them +- `sc.retention.consecutive_failures_total` - Consecutive failures of that pass + - `retention.entity` - Which pass is failing + +### Reading the retention metrics + +- Rows reclaimed per hour: `sum(rate(sc_retention_rows_deleted_total[1h])) by (exported_job,retention_entity)` +- Sweep duration: `histogram_quantile(0.9,sum(rate(sc_retention_cycle_duration_seconds_bucket[6h])) by (le,exported_job,retention_entity))` +- Retention is broken: `max(sc_retention_consecutive_failures_total) by (exported_job,retention_entity) > 2` + +What the shapes mean: + +- `consecutive_failures_total` above zero is the signal that rows are no longer being reclaimed. Nothing else in the product reports this, and the database grows without bound while it lasts. +- `rows_deleted_total` flat at zero across a long window is only healthy if the instance is also not ingesting. Deletion stopping while ingestion continues means the retention window is not being enforced. +- A cycle duration climbing towards the hourly interval means the sweep is no longer keeping up with the arrival rate, and each run starts further behind than the last. +- A body store that refuses a delete fails the `failed_messages` pass rather than orphaning the body, so an expired credential or a changed permission shows as a climbing gauge instead of storage that quietly keeps growing. Those rows stay until a sweep can delete the body and the row together. +- Each pass is isolated, so one kind of row failing to be reclaimed does not stop the others. Every pass reports a result on every run, and the gauge is per pass, so alert across all entities rather than on any one of them. + ## Monitoring No telemetry is currently available. From fae750c9492c0422cf9f9d0e69c686a039716a08 Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 31 Aug 2026 15:23:51 +1000 Subject: [PATCH 3/7] Rename DeleteBody to DeleteBodyIfExists on IBodyStoragePersistence --- .../BodyStorage/AzureBlobBodyStoragePersistence.cs | 2 +- .../BodyStorage/FileSystemBodyStoragePersistence.cs | 2 +- .../Implementation/BodyStorage/S3BodyStoragePersistence.cs | 2 +- .../Implementation/FailedErrorImportDataStore.cs | 2 +- .../Infrastructure/IBodyStoragePersistence.cs | 7 ++++++- .../EFCore/AzureBlobBodyStorageTests.cs | 4 ++-- .../EFCore/BodyStoragePersistenceTests.cs | 4 ++-- .../EFCore/InMemoryBodyStoragePersistence.cs | 4 ++-- .../EFCore/S3BodyStorageTests.cs | 4 ++-- 9 files changed, 18 insertions(+), 13 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStoragePersistence.cs index 7744152903..f07ab437a7 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStoragePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStoragePersistence.cs @@ -93,6 +93,6 @@ public async Task WriteBody(string bodyId, ReadOnlyMemory body, string con } } - public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) => + public Task DeleteBodyIfExists(string bodyId, CancellationToken cancellationToken = default) => container.GetBlobClient(bodyId).DeleteIfExistsAsync(cancellationToken: cancellationToken); } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStoragePersistence.cs index 6f5ba321c7..a9ed66165b 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStoragePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStoragePersistence.cs @@ -135,7 +135,7 @@ public async Task WriteBody(string bodyId, ReadOnlyMemory body, string con } } - public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) + public Task DeleteBodyIfExists(string bodyId, CancellationToken cancellationToken = default) { TryDelete(GetBodyFilePath(bodyId)); return Task.CompletedTask; diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStoragePersistence.cs index a6065177e4..f5194dab85 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStoragePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStoragePersistence.cs @@ -93,7 +93,7 @@ public async Task WriteBody(string bodyId, ReadOnlyMemory body, string con } } - public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) => + public Task DeleteBodyIfExists(string bodyId, CancellationToken cancellationToken = default) => client.DeleteObjectAsync(bucketName, Key(bodyId), cancellationToken); async Task Exists(string key, CancellationToken cancellationToken) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs index a47894baad..547e5e4678 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs @@ -189,7 +189,7 @@ async Task DeleteExternalBody(Guid uniqueMessageId, CancellationToken cancellati { try { - await bodyStorage.DeleteBody(FailedErrorImportEntity.ExternalBodyId(uniqueMessageId), cancellationToken); + await bodyStorage.DeleteBodyIfExists(FailedErrorImportEntity.ExternalBodyId(uniqueMessageId), cancellationToken); } #pragma warning disable PS0019 // The filter already excludes OperationCanceledException, so cancellation // propagates; PS0019 only recognises a cancellationToken.IsCancellationRequested guard. diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs index 2427951fae..685efe7a01 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs @@ -15,5 +15,10 @@ public interface IBodyStoragePersistence { Task WriteBody(string bodyId, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default); Task ReadBody(string bodyId, CancellationToken cancellationToken = default); - Task DeleteBody(string bodyId, CancellationToken cancellationToken = default); + + /// + /// Implementations throw only when the store itself fails, never because the body was already + /// gone: callers delete the body before the row that names it. + /// + Task DeleteBodyIfExists(string bodyId, CancellationToken cancellationToken = default); } diff --git a/src/ServiceControl.Persistence.Tests/EFCore/AzureBlobBodyStorageTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/AzureBlobBodyStorageTests.cs index 12ffdc16cc..6f8ec73091 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/AzureBlobBodyStorageTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/AzureBlobBodyStorageTests.cs @@ -108,7 +108,7 @@ public async Task Delete_removes_the_body() var bodyId = Guid.NewGuid().ToString(); await store.WriteBody(bodyId, "payload"u8.ToArray(), "text/plain"); - await store.DeleteBody(bodyId); + await store.DeleteBodyIfExists(bodyId); Assert.That(await store.ReadBody(bodyId), Is.Null); } @@ -117,7 +117,7 @@ public async Task Delete_removes_the_body() public async Task Delete_of_a_missing_body_does_not_throw() { var store = await CreateContainer(); - Assert.DoesNotThrowAsync(() => store.DeleteBody(Guid.NewGuid().ToString())); + Assert.DoesNotThrowAsync(() => store.DeleteBodyIfExists(Guid.NewGuid().ToString())); } [Test] diff --git a/src/ServiceControl.Persistence.Tests/EFCore/BodyStoragePersistenceTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/BodyStoragePersistenceTests.cs index 357289ddff..070056f6d1 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/BodyStoragePersistenceTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/BodyStoragePersistenceTests.cs @@ -107,7 +107,7 @@ public async Task Delete_removes_the_body(string kind) var bodyId = Guid.NewGuid().ToString(); await store.WriteBody(bodyId, Encoding.UTF8.GetBytes("payload"), "text/plain"); - await store.DeleteBody(bodyId); + await store.DeleteBodyIfExists(bodyId); Assert.That(await store.ReadBody(bodyId), Is.Null); } @@ -118,7 +118,7 @@ public void Delete_of_a_missing_body_does_not_throw(string kind) { var store = CreateStore(kind); - Assert.DoesNotThrowAsync(() => store.DeleteBody(Guid.NewGuid().ToString())); + Assert.DoesNotThrowAsync(() => store.DeleteBodyIfExists(Guid.NewGuid().ToString())); } [TestCase(InMemory)] diff --git a/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs b/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs index 4190e37f58..4c0d30b2c2 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs @@ -79,11 +79,11 @@ public void Evict(string bodyId) } } - public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) + public Task DeleteBodyIfExists(string bodyId, CancellationToken cancellationToken = default) { if (FailDeleteFor.Contains(bodyId)) { - throw new InvalidOperationException($"Simulated missing body for {bodyId}"); + throw new InvalidOperationException($"Simulated body storage failure for {bodyId}"); } lock (gate) diff --git a/src/ServiceControl.Persistence.Tests/EFCore/S3BodyStorageTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/S3BodyStorageTests.cs index e820e9b814..1dedb815e1 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/S3BodyStorageTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/S3BodyStorageTests.cs @@ -105,7 +105,7 @@ public async Task Delete_removes_the_body() var bodyId = Guid.NewGuid().ToString(); await store.WriteBody(bodyId, Encoding.UTF8.GetBytes("payload"), "text/plain"); - await store.DeleteBody(bodyId); + await store.DeleteBodyIfExists(bodyId); Assert.That(await store.ReadBody(bodyId), Is.Null); } @@ -114,7 +114,7 @@ public async Task Delete_removes_the_body() public async Task Delete_of_a_missing_body_does_not_throw() { var store = await CreateBucket(); - Assert.DoesNotThrowAsync(() => store.DeleteBody(Guid.NewGuid().ToString())); + Assert.DoesNotThrowAsync(() => store.DeleteBodyIfExists(Guid.NewGuid().ToString())); } [Test] From f87b10fac826e2510aba0a61b0a2bb2d5aae3395 Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 31 Aug 2026 15:23:51 +1000 Subject: [PATCH 4/7] Move metrics types into Infrastructure/Metrics sub-namespace --- .../Abstractions/BasePersistence.cs | 1 + .../Infrastructure/Metrics/CycleOutcome.cs | 8 +++ .../Metrics/RetentionCycleMetrics.cs | 40 ++++++++++++ .../Infrastructure/Metrics/RetentionEntity.cs | 8 +++ .../{ => Metrics}/RetentionMetrics.cs | 62 +++++-------------- .../Infrastructure/RetentionSweeper.cs | 1 + .../EFCore/RecordedRetentionMetrics.cs | 2 +- .../EFCore/RetentionMetricsTests.cs | 2 +- .../EFCore/RetentionSweepTests.cs | 2 +- 9 files changed, 77 insertions(+), 49 deletions(-) create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/CycleOutcome.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionCycleMetrics.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionEntity.cs rename src/ServiceControl.Persistence.EFCore/Infrastructure/{ => Metrics}/RetentionMetrics.cs (70%) diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index 4a0195210d..6988f622d5 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -12,6 +12,7 @@ namespace ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.Implementation.Recoverability; using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; using ServiceControl.Persistence.EFCore.Infrastructure; +using ServiceControl.Persistence.EFCore.Infrastructure.Metrics; using ServiceControl.Persistence.MessageRedirects; using ServiceControl.Persistence.Recoverability; using ServiceControl.Persistence.UnitOfWork; diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/CycleOutcome.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/CycleOutcome.cs new file mode 100644 index 0000000000..ac1c6d0030 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/CycleOutcome.cs @@ -0,0 +1,8 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure.Metrics; + +enum CycleOutcome +{ + Success, + Cancelled, + Failed +} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionCycleMetrics.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionCycleMetrics.cs new file mode 100644 index 0000000000..d051aada3d --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionCycleMetrics.cs @@ -0,0 +1,40 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure.Metrics; + +using System.Diagnostics; + +/// +/// One pass of the retention sweep. A pass that finished counts as a success even if shutdown has +/// since been requested; one that shutdown cut short is recorded as cancelled rather than as a +/// failure. +/// +public sealed class RetentionCycleMetrics : IDisposable +{ + internal RetentionCycleMetrics(RetentionMetrics metrics, RetentionEntity entity, CancellationToken cancellationToken) + { + this.metrics = metrics; + this.entity = entity; + this.cancellationToken = cancellationToken; + } + + // The pass is over once it completes, so the clock stops here rather than wherever the scope + // happens to be disposed. + public void Complete() + { + stopwatch.Stop(); + completed = true; + } + + public void Dispose() => metrics.RecordCycle(entity, stopwatch.Elapsed, Outcome); + + CycleOutcome Outcome => + completed ? CycleOutcome.Success + : cancellationToken.IsCancellationRequested ? CycleOutcome.Cancelled + : CycleOutcome.Failed; + + bool completed; + + readonly RetentionMetrics metrics; + readonly RetentionEntity entity; + readonly CancellationToken cancellationToken; + readonly Stopwatch stopwatch = Stopwatch.StartNew(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionEntity.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionEntity.cs new file mode 100644 index 0000000000..5fcd43f587 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionEntity.cs @@ -0,0 +1,8 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure.Metrics; + +public enum RetentionEntity +{ + FailedMessages, + EventLog, + GroupComments +} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionMetrics.cs similarity index 70% rename from src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs rename to src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionMetrics.cs index 34e384de7b..08f1f97f64 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/Metrics/RetentionMetrics.cs @@ -1,16 +1,9 @@ -namespace ServiceControl.Persistence.EFCore.Infrastructure; +namespace ServiceControl.Persistence.EFCore.Infrastructure.Metrics; using System.Diagnostics; using System.Diagnostics.Metrics; using ServiceControl.Infrastructure; -public enum RetentionEntity -{ - FailedMessages, - EventLog, - GroupComments -} - public class RetentionMetrics { public const string MeterName = ServiceControlMeters.Error; @@ -23,7 +16,7 @@ public RetentionMetrics(IMeterFactory meterFactory) { var meter = meterFactory.Create(MeterName, MeterVersion); - cycleDuration = meter.CreateHistogram( + cycleDuration = meter.CreateHistogram( CycleDurationInstrumentName, unit: "seconds", description: "Retention sweep pass duration in seconds", @@ -40,18 +33,20 @@ public RetentionMetrics(IMeterFactory meterFactory) public void RecordRowsDeleted(RetentionEntity entity, int rows) => rowsDeleted.Add(rows, EntityTags[(int)entity]); - internal void RecordCycle(RetentionEntity entity, TimeSpan elapsed, bool success) + internal void RecordCycle(RetentionEntity entity, TimeSpan elapsed, CycleOutcome outcome) { var tags = EntityTags[(int)entity]; - tags.Add("result", success ? "success" : "failed"); + tags.Add("result", ResultTag(outcome)); cycleDuration.Record(elapsed.TotalSeconds, tags); - if (success) + // A pass cut short by shutdown neither proves the sweep healthy nor faulty, so it leaves + // the gauge where it was. + if (outcome == CycleOutcome.Success) { Interlocked.Exchange(ref consecutiveFailures[(int)entity], 0); } - else + else if (outcome == CycleOutcome.Failed) { Interlocked.Increment(ref consecutiveFailures[(int)entity]); } @@ -67,6 +62,14 @@ IEnumerable> ObserveConsecutiveFailures() static TagList EntityTag(string entity) => new() { { "retention.entity", entity } }; + static string ResultTag(CycleOutcome outcome) => outcome switch + { + CycleOutcome.Success => "success", + CycleOutcome.Cancelled => "cancelled", + CycleOutcome.Failed => "failed", + _ => throw new ArgumentOutOfRangeException(nameof(outcome)) + }; + readonly long[] consecutiveFailures = new long[EntityTags.Length]; readonly Histogram cycleDuration; @@ -85,36 +88,3 @@ IEnumerable> ObserveConsecutiveFailures() const string MeterVersion = "0.1.0"; const string InstrumentPrefix = "sc.retention"; } - -/// -/// One pass of the retention sweep. A pass interrupted by shutdown is not a measurement of -/// anything, so a cancelled cycle records neither a duration nor a failure. -/// -public sealed class RetentionCycleMetrics : IDisposable -{ - internal RetentionCycleMetrics(RetentionMetrics metrics, RetentionEntity entity, CancellationToken cancellationToken) - { - this.metrics = metrics; - this.entity = entity; - this.cancellationToken = cancellationToken; - } - - public void Complete() => completed = true; - - public void Dispose() - { - if (cancellationToken.IsCancellationRequested) - { - return; - } - - metrics.RecordCycle(entity, stopwatch.Elapsed, completed); - } - - bool completed; - - readonly RetentionMetrics metrics; - readonly RetentionEntity entity; - readonly CancellationToken cancellationToken; - readonly Stopwatch stopwatch = Stopwatch.StartNew(); -} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs index fd57ae2faa..eda88a86b8 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs @@ -8,6 +8,7 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.DbContexts; using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure.Metrics; // Deletes rows once they age past their retention period. // Runs hourly, in bounded batches so it never holds a large delete, and recomputes the cutoffs on diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs b/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs index 687164d117..903d0c4bf8 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs @@ -4,7 +4,7 @@ namespace ServiceControl.Persistence.Tests; using System.Collections.Generic; using System.Diagnostics.Metrics; using System.Linq; -using ServiceControl.Persistence.EFCore.Infrastructure; +using ServiceControl.Persistence.EFCore.Infrastructure.Metrics; /// /// Collects everything the retention instruments record, for the meter belonging to one factory. diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs index 774f5a8a0a..6bbf960819 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs @@ -8,7 +8,7 @@ namespace ServiceControl.Persistence.Tests; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using NUnit.Framework; -using ServiceControl.Persistence.EFCore.Infrastructure; +using ServiceControl.Persistence.EFCore.Infrastructure.Metrics; /// /// Instrument names are what dashboards and alerts are built on, so they are a published contract diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs index 8948999fe6..daa9d20e78 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs @@ -10,7 +10,7 @@ namespace ServiceControl.Persistence.Tests; using ServiceControl.EventLog; using ServiceControl.MessageFailures; using ServiceControl.Persistence.EFCore.Entities; -using ServiceControl.Persistence.EFCore.Infrastructure; +using ServiceControl.Persistence.EFCore.Infrastructure.Metrics; using ServiceControl.Persistence.Infrastructure; class RetentionSweepTests : ErrorIngestionTestBase From cd2c458466ea1de5d1c7cae6ea294d3efb20a026 Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 31 Aug 2026 15:23:51 +1000 Subject: [PATCH 5/7] Record cancelled retention cycles instead of silently dropping them --- .../EFCore/RetentionMetricsTests.cs | 38 ++++++++++++++++++- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs index 6bbf960819..5caf1feacd 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs @@ -130,7 +130,7 @@ public void A_success_clears_the_failures_of_that_entity_alone() } [Test] - public void A_cycle_interrupted_by_shutdown_is_not_recorded() + public void A_cycle_interrupted_by_shutdown_is_recorded_as_cancelled() { var metrics = new RetentionMetrics(MeterFactory); using var recorded = Listen(); @@ -141,13 +141,47 @@ public void A_cycle_interrupted_by_shutdown_is_not_recorded() shutdown.Cancel(); } + var cycles = recorded.Cycles(RetentionEntity.FailedMessages); + using (Assert.EnterMultipleScope()) { - Assert.That(recorded.Cycles(RetentionEntity.FailedMessages), Is.Empty); + Assert.That(cycles, Has.Count.EqualTo(1)); + Assert.That(cycles[0].Result, Is.EqualTo("cancelled")); Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.Zero); } } + [Test] + public void A_cycle_that_finished_before_shutdown_is_still_a_success() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + using var shutdown = new CancellationTokenSource(); + + using (var cycle = metrics.BeginCycle(RetentionEntity.FailedMessages, shutdown.Token)) + { + cycle.Complete(); + shutdown.Cancel(); + } + + Assert.That(recorded.Cycles(RetentionEntity.FailedMessages)[0].Result, Is.EqualTo("success")); + } + + [Test] + public void Shutdown_neither_clears_nor_adds_to_the_failures() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + using var shutdown = new CancellationTokenSource(); + + metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose(); + + shutdown.Cancel(); + metrics.BeginCycle(RetentionEntity.FailedMessages, shutdown.Token).Dispose(); + + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.EqualTo(1)); + } + [Test] public void Deleted_rows_are_counted_per_entity() { From 04dca951ce61e9160f2d201b2b144ad65e5632a9 Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 31 Aug 2026 15:23:51 +1000 Subject: [PATCH 6/7] Isolate retention passes so one failure does not block the others, and propagate body-delete failures --- .../Infrastructure/RetentionSweeper.cs | 65 +++++++++---------- 1 file changed, 32 insertions(+), 33 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs index eda88a86b8..5396e060da 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs @@ -63,9 +63,30 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken = async Task Sweep(bool pace, CancellationToken cancellationToken) { - await SweepFailedMessages(pace, cancellationToken); - await SweepEventLogItems(pace, cancellationToken); - await SweepOrphanedGroupComments(cancellationToken); + await RunPass(RetentionEntity.FailedMessages, token => SweepFailedMessages(pace, token), cancellationToken); + await RunPass(RetentionEntity.EventLog, token => SweepEventLogItems(pace, token), cancellationToken); + await RunPass(RetentionEntity.GroupComments, SweepOrphanedGroupComments, cancellationToken); + } + + // Each pass is isolated so one failing kind of row does not stop the others from being + // reclaimed, and so the metrics report an outcome for every pass on every run. + async Task RunPass(RetentionEntity entity, Func pass, CancellationToken cancellationToken) + { + using var cycle = metrics.BeginCycle(entity, cancellationToken); + + try + { + await pass(cancellationToken); + + cycle.Complete(); + } +#pragma warning disable PS0019 // The filter already excludes OperationCanceledException, so + // cancellation propagates; PS0019 only recognises a cancellationToken guard. + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogError(ex, "Error during the {RetentionEntity} retention pass", entity); + } +#pragma warning restore PS0019 } // Once the last message of a group has been swept the group cannot be displayed at all, so its @@ -73,7 +94,6 @@ async Task Sweep(bool pace, CancellationToken cancellationToken) // group ids are deterministic, reattach a stale comment if the same failure ever recurs. async Task SweepOrphanedGroupComments(CancellationToken cancellationToken) { - using var cycle = metrics.BeginCycle(RetentionEntity.GroupComments, cancellationToken); using var scope = serviceScopeFactory.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); @@ -82,16 +102,12 @@ async Task SweepOrphanedGroupComments(CancellationToken cancellationToken) .ExecuteDeleteAsync(cancellationToken); metrics.RecordRowsDeleted(RetentionEntity.GroupComments, deleted); - - cycle.Complete(); } // Event log items are insert-only and carry no external bodies, so each batch is a single // ordered DELETE. async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) { - using var cycle = metrics.BeginCycle(RetentionEntity.EventLog, cancellationToken); - var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.EventsRetentionPeriod; while (!cancellationToken.IsCancellationRequested) @@ -117,14 +133,10 @@ async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) await Task.Delay(BatchPause, timeProvider, cancellationToken); } } - - cycle.Complete(); } async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) { - using var cycle = metrics.BeginCycle(RetentionEntity.FailedMessages, cancellationToken); - var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.ErrorRetentionPeriod; while (!cancellationToken.IsCancellationRequested) @@ -142,15 +154,19 @@ async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) if (expired.Count == 0) { + // The other two passes always report what their delete removed, so this one + // reports its zero rather than leaving a gap in the series. + metrics.RecordRowsDeleted(RetentionEntity.FailedMessages, 0); break; } - // External bodies are deleted before the rows. A crash in between leaves rows the next - // sweep re-handles (tolerating the already-missing body); deleting rows first would - // instead leak the external bodies. + // External bodies are deleted before the rows, so a body that will not delete fails the + // pass with its row intact and the next sweep retries it. Every store already treats an + // already-missing body as a success, so anything reaching here is a storage failure and + // deleting the row would strand the body with nothing left to name it. foreach (var row in expired.Where(row => row.BodyStoredExternally)) { - await DeleteExternalBody(row.UniqueMessageId, cancellationToken); + await bodyStorage.DeleteBodyIfExists(row.UniqueMessageId.ToString(), cancellationToken); } var ids = expired.Select(row => row.UniqueMessageId).ToArray(); @@ -174,23 +190,6 @@ async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) await Task.Delay(BatchPause, timeProvider, cancellationToken); } } - - cycle.Complete(); - } - - async Task DeleteExternalBody(Guid uniqueMessageId, CancellationToken cancellationToken) - { - try - { - await bodyStorage.DeleteBody(uniqueMessageId.ToString(), cancellationToken); - } -#pragma warning disable PS0019 // As above: the filter excludes cancellation already. - catch (Exception ex) when (ex is not OperationCanceledException) - { - // Retention must not stall on a missing or unavailable body. - logger.LogWarning(ex, "Could not delete the external body for {UniqueMessageId} during retention", uniqueMessageId); - } -#pragma warning restore PS0019 } static System.Linq.Expressions.Expression> IsExpired(DateTime cutoff) => From 877454bc66b5d6c676d35f1d4e60d8aa3cf3f98d Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 31 Aug 2026 15:23:51 +1000 Subject: [PATCH 7/7] Add and update tests for revised retention sweep behaviour --- .../EFCore/RetentionSweepTests.cs | 69 +++++++++++++++++-- 1 file changed, 64 insertions(+), 5 deletions(-) diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs index daa9d20e78..761e624e12 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs @@ -105,20 +105,39 @@ public async Task Also_removes_the_group_rows_of_swept_messages() } [Test] - public async Task Tolerates_a_body_that_cannot_be_deleted() + public async Task Keeps_the_row_of_a_body_that_cannot_be_deleted() { var unluckyBody = await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-31), bodyStoredExternally: true); - var otherBody = await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-31), bodyStoredExternally: true); RecordedBodies.FailDeleteFor.Add(unluckyBody.ToString()); + using var recorded = ListenToRetentionMetrics(); + + await RunRetentionSweep(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(await FindFailedMessage(unluckyBody), Is.Not.Null, "deleting the row would leave the body with nothing to name it"); + Assert.That(recorded.Cycles(RetentionEntity.FailedMessages).Select(cycle => cycle.Result), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.Cycles(RetentionEntity.EventLog).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + } + } + + [Test] + public async Task Retries_a_body_that_could_not_be_deleted_on_the_next_sweep() + { + var unluckyBody = await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-31), bodyStoredExternally: true); + RecordedBodies.FailDeleteFor.Add(unluckyBody.ToString()); + + await RunRetentionSweep(); + + RecordedBodies.FailDeleteFor.Clear(); + await RunRetentionSweep(); using (Assert.EnterMultipleScope()) { - // The failed delete must not stall retention: both rows are still swept. Assert.That(await FindFailedMessage(unluckyBody), Is.Null); - Assert.That(await FindFailedMessage(otherBody), Is.Null); - Assert.That(RecordedBodies.Deleted, Does.Contain(otherBody.ToString())); + Assert.That(RecordedBodies.Deleted, Does.Contain(unluckyBody.ToString())); } } @@ -211,6 +230,46 @@ public async Task Records_a_successful_cycle_for_every_pass() } } + [Test] + public async Task Reports_zero_deleted_rows_when_there_is_nothing_to_sweep() + { + using var recorded = ListenToRetentionMetrics(); + + await RunRetentionSweep(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.Of(RetentionMetrics.RowsDeletedInstrumentName, RetentionEntity.FailedMessages), Is.Not.Empty); + Assert.That(recorded.Of(RetentionMetrics.RowsDeletedInstrumentName, RetentionEntity.EventLog), Is.Not.Empty); + Assert.That(recorded.Of(RetentionMetrics.RowsDeletedInstrumentName, RetentionEntity.GroupComments), Is.Not.Empty); + Assert.That(recorded.RowsDeleted(RetentionEntity.FailedMessages), Is.Zero); + } + } + + [Test] + public async Task A_failing_pass_does_not_stop_the_others() + { + EFSettings.EventsRetentionPeriod = TimeSpan.FromDays(14); + await Store(EventLogRow("expired", Now.AddDays(-15))); + + // Subtracting this from the clock cannot be represented, so the failed messages pass throws + // before it reaches the database. + EFSettings.ErrorRetentionPeriod = TimeSpan.FromDays(1_000_000); + + using var recorded = ListenToRetentionMetrics(); + + await RunRetentionSweep(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.Cycles(RetentionEntity.FailedMessages).Select(cycle => cycle.Result), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.EqualTo(1)); + Assert.That(recorded.Cycles(RetentionEntity.EventLog).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.Cycles(RetentionEntity.GroupComments).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + Assert.That(await GetRemainingMarkers(), Does.Not.Contain("expired")); + } + } + RecordedRetentionMetrics ListenToRetentionMetrics() => new(ServiceProvider.GetRequiredService()); async Task SeedGroup(Guid uniqueMessageId)