Add custom check for failing retention

src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs

@@ -47,6 +47,9 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste
             services.AddSingleton<RetentionSweeper>();
             services.AddHostedService(sp => sp.GetRequiredService<RetentionSweeper>());
             services.AddSingleton<IRetentionSweeper>(sp => sp.GetRequiredService<RetentionSweeper>());
+            services.AddSingleton<IRetentionSweepHealth>(sp => sp.GetRequiredService<RetentionSweeper>());
+
+            services.AddCustomCheck<RetentionSweepCustomCheck>();
         }
 
         services.AddSingleton<OperationsManager>();

src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweepCustomCheck.cs

@@ -0,0 +1,25 @@
+namespace ServiceControl.Persistence.EFCore.Infrastructure;
+
+using System.Threading;
+using System.Threading.Tasks;
+using NServiceBus.CustomChecks;
+
+interface IRetentionSweepHealth
+{
+    string? GetFailureSummary();
+}
+
+// Fails after any retention pass fails and recovers after the next fully successful sweep.
+// The standard custom-check state-transition pipeline deduplicates repeated results.
+class RetentionSweepCustomCheck(IRetentionSweepHealth retentionSweepHealth)
+    : CustomCheck("ServiceControl Retention", "ServiceControl Health", TimeSpan.FromMinutes(1))
+{
+    public override Task<CheckResult> PerformCheck(CancellationToken cancellationToken = default)
+    {
+        var failureSummary = retentionSweepHealth.GetFailureSummary();
+
+        return Task.FromResult(failureSummary is null
+            ? CheckResult.Pass
+            : CheckResult.Failed($"Retention processing has failures. Last failure per entity: {failureSummary}"));
+    }
+}

src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs

@@ -24,13 +24,15 @@ public class RetentionSweeper(
     IBodyStoragePersistence bodyStorage,
     RetentionMetrics metrics,
     EFPersisterSettings settings,
-    IHostApplicationLifetime hostApplicationLifetime) : BackgroundService, IRetentionSweeper
+    IHostApplicationLifetime hostApplicationLifetime) : BackgroundService, IRetentionSweeper, IRetentionSweepHealth
 {
     const int BatchSize = 1000;
     static readonly TimeSpan Interval = TimeSpan.FromHours(1);
     static readonly TimeSpan InitialDelay = TimeSpan.FromMinutes(1);
     static readonly TimeSpan BatchPause = TimeSpan.FromSeconds(1);
 
+    readonly Dictionary<RetentionEntity, string> failures = [];
+
     // Single-flight guard shared by the hourly timer path and the manual API path so two sweeps
     // never overlap. Precedent: ExternalIntegrationRequestsDataStore.drainLock.
     readonly SemaphoreSlim sweepLock = new(1, 1);
@@ -129,6 +131,32 @@ async Task SweepWithoutAcquiringLock()
 
     public RetentionSweepCurrentStatus GetStatus() => new(isRunning, lastStartedAt, lastFinishedAt, lastErrorCutoff, lastEventsCutoff);
 
+    internal void RecordFailure(RetentionEntity entity, string reason)
+    {
+        lock (failures)
+        {
+            failures[entity] = reason;
+        }
+    }
+
+    internal (RetentionEntity Entity, string Reason)[] GetActiveFailures()
+    {
+        lock (failures)
+        {
+            return failures.Select(failure => (failure.Key, failure.Value)).ToArray();
+        }
+    }
+
+    string? IRetentionSweepHealth.GetFailureSummary()
+    {
+        lock (failures)
+        {
+            return failures.Count == 0
+                ? null
+                : string.Join("; ", failures.Select(failure => $"{failure.Key}: {failure.Value}"));
+        }
+    }
+
     async Task Sweep(DateTime? errorCutoff, DateTime? eventsCutoff, bool pace, CancellationToken cancellationToken)
     {
         await sweepLock.WaitAsync(cancellationToken);
@@ -152,28 +180,38 @@ async Task Sweep(DateTime? errorCutoff, DateTime? eventsCutoff, bool pace, Cance
     // manual background path (which already holds the lock) share one implementation.
     async Task SweepBody(DateTime? errorCutoff, DateTime? eventsCutoff, bool pace, CancellationToken cancellationToken)
     {
-        await RunPass(RetentionEntity.FailedMessages, token => SweepFailedMessages(pace, errorCutoff, token), cancellationToken);
-        await RunPass(RetentionEntity.EventLog, token => SweepEventLogItems(pace, eventsCutoff, token), cancellationToken);
-        await RunPass(RetentionEntity.GroupComments, SweepOrphanedGroupComments, cancellationToken);
+        var failedMessagesSucceeded = await RunPass(RetentionEntity.FailedMessages, token => SweepFailedMessages(pace, errorCutoff, token), cancellationToken);
+        var eventLogSucceeded = await RunPass(RetentionEntity.EventLog, token => SweepEventLogItems(pace, eventsCutoff, token), cancellationToken);
+        var groupCommentsSucceeded = await RunPass(RetentionEntity.GroupComments, SweepOrphanedGroupComments, cancellationToken);
+
+        if (failedMessagesSucceeded && eventLogSucceeded && groupCommentsSucceeded)
+        {
+            lock (failures)
+            {
+                failures.Clear();
+            }
+        }
     }
 
     // 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<CancellationToken, Task> pass, CancellationToken cancellationToken)
+    async Task<bool> RunPass(RetentionEntity entity, Func<CancellationToken, Task> pass, CancellationToken cancellationToken)
     {
         using var cycle = metrics.BeginCycle(entity, cancellationToken);
 
         try
         {
             await pass(cancellationToken);
-
             cycle.Complete();
+            return true;
         }
 #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);
+            RecordFailure(entity, ex.Message);
+            return false;
         }
 #pragma warning restore PS0019
     }

src/ServiceControl.Persistence.Tests.PostgreSql/ApprovalFiles/CustomCheckTests.VerifyCustomChecks.approved.txt

@@ -1 +1,2 @@
+ServiceControl Health: ServiceControl Retention
 Storage space: ServiceControl body storage

src/ServiceControl.Persistence.Tests.SqlServer/ApprovalFiles/CustomCheckTests.VerifyCustomChecks.approved.txt

@@ -1 +1,2 @@
+ServiceControl Health: ServiceControl Retention
 Storage space: ServiceControl body storage

src/ServiceControl.Persistence.Tests/EFCore/FileSystemBodyStorageCustomCheckTests.cs

@@ -4,6 +4,7 @@ namespace ServiceControl.Persistence.Tests;
 using System.IO;
 using System.Linq;
 using System.Threading.Tasks;
+using EFCore.Infrastructure;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging.Abstractions;
 using NServiceBus.CustomChecks;
@@ -136,6 +137,7 @@ public void Is_registered_and_resolvable_for_file_system_body_storage()
             ConnectionString = "Server=nowhere",
             BodyStorage = CreateSettings(StoragePath, 15)
         });
+        services.AddSingleton<IRetentionSweepHealth, StubRetentionHealth>();
 
         var check = services.BuildServiceProvider().GetServices<ICustomCheck>().OfType<FileSystemBodyStorageCustomCheck>().SingleOrDefault();
 
@@ -196,4 +198,8 @@ class ThrowingDriveSpaceProvider(Exception exception) : IDriveSpaceProvider
     }
 
     sealed class TestPersisterSettings : EFPersisterSettings;
+    sealed class StubRetentionHealth : IRetentionSweepHealth
+    {
+        public string GetFailureSummary() => "";
+    }
 }

src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepCustomCheckTests.cs

@@ -0,0 +1,113 @@
+namespace ServiceControl.Persistence.Tests;
+
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using NServiceBus.CustomChecks;
+using NUnit.Framework;
+using ServiceControl.Contracts.CustomChecks;
+using ServiceControl.CustomChecks;
+using ServiceControl.Infrastructure.DomainEvents;
+using ServiceControl.Operations;
+using ServiceControl.Persistence.EFCore.Infrastructure;
+using ServiceControl.Persistence.EFCore.Infrastructure.Metrics;
+
+class RetentionSweepCustomCheckTests : PersistenceTestBase
+{
+    RetentionSweeper Sweeper => ServiceProvider.GetRequiredService<RetentionSweeper>();
+
+    RetentionSweepCustomCheck Check =>
+        ServiceProvider.GetServices<ICustomCheck>().OfType<RetentionSweepCustomCheck>().Single();
+
+    [Test]
+    public async Task Check_initially_passes()
+    {
+        var result = await Check.PerformCheck();
+        Assert.That(result, Is.EqualTo(CheckResult.Pass));
+    }
+
+    [Test]
+    public async Task Check_fails_after_a_retention_failure()
+    {
+        Sweeper.RecordFailure(RetentionEntity.FailedMessages, "db timeout");
+
+        var result = await Check.PerformCheck();
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(result.HasFailed, Is.True);
+            Assert.That(result.FailureReason, Does.Contain("FailedMessages"));
+            Assert.That(result.FailureReason, Does.Contain("db timeout"));
+        }
+    }
+
+    [Test]
+    public async Task Check_passes_after_a_fully_successful_sweep()
+    {
+        Sweeper.RecordFailure(RetentionEntity.FailedMessages, "db timeout");
+
+        await Sweeper.SweepNow();
+
+        var result = await Check.PerformCheck();
+        Assert.That(result, Is.EqualTo(CheckResult.Pass));
+    }
+
+    [Test]
+    public async Task Check_does_not_expire_with_time()
+    {
+        Sweeper.RecordFailure(RetentionEntity.FailedMessages, "db timeout");
+        AdvanceClock(TimeSpan.FromHours(2));
+
+        var result = await Check.PerformCheck();
+        Assert.That(result.HasFailed, Is.True);
+    }
+
+    [Test]
+    public async Task Multiple_failing_entities_are_represented()
+    {
+        Sweeper.RecordFailure(RetentionEntity.FailedMessages, "body delete failed");
+        Sweeper.RecordFailure(RetentionEntity.EventLog, "batch delete timeout");
+
+        var result = await Check.PerformCheck();
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(result.HasFailed, Is.True);
+            Assert.That(result.FailureReason, Does.Contain("FailedMessages"));
+            Assert.That(result.FailureReason, Does.Contain("EventLog"));
+            Assert.That(result.FailureReason, Does.Contain("body delete failed"));
+            Assert.That(result.FailureReason, Does.Contain("batch delete timeout"));
+        }
+    }
+
+    [Test]
+    public async Task Repeated_failed_results_raise_one_state_change_event()
+    {
+        Sweeper.RecordFailure(RetentionEntity.FailedMessages, "db timeout");
+
+        var result = await Check.PerformCheck();
+        var domainEvents = (FakeDomainEvents)ServiceProvider.GetRequiredService<IDomainEvents>();
+        var processor = new CustomCheckResultProcessor(domainEvents, CustomChecks, NullLogger<CustomCheckResultProcessor>.Instance);
+        var detail = new CustomCheckDetail
+        {
+            Category = "ServiceControl Health",
+            CustomCheckId = "ServiceControl Retention",
+            HasFailed = result.HasFailed,
+            FailureReason = result.FailureReason,
+            ReportedAt = Now,
+            OriginatingEndpoint = new EndpointDetails
+            {
+                Host = "localhost",
+                HostId = Guid.NewGuid(),
+                Name = "ServiceControl"
+            }
+        };
+
+        await processor.ProcessResult(detail);
+        await processor.ProcessResult(detail);
+
+        Assert.That(domainEvents.RaisedEvents.OfType<CustomCheckFailed>().Count(), Is.EqualTo(1));
+    }
+}

src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs

@@ -10,6 +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;
 
@@ -308,6 +309,24 @@ public async Task A_failing_pass_does_not_stop_the_others()
         }
     }
 
+    [Test]
+    public async Task A_failing_pass_is_tracked()
+    {
+        // 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);
+
+        await RunRetentionSweep();
+
+        var failure = ServiceProvider.GetRequiredService<RetentionSweeper>().GetActiveFailures().Single();
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(failure.Entity, Is.EqualTo(RetentionEntity.FailedMessages));
+            Assert.That(failure.Reason, Is.Not.Empty);
+        }
+    }
+
     RecordedRetentionMetrics ListenToRetentionMetrics() => new(ServiceProvider.GetRequiredService<IMeterFactory>());
 
     async Task<string> SeedGroup(Guid uniqueMessageId)

src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs

@@ -34,6 +34,7 @@ public async Task SetUp()
         }
 
         var hostBuilder = Host.CreateApplicationBuilder();
+        hostBuilder.Services.AddMetrics();
 
         LoggerUtil.ActiveLoggers = Loggers.Test;
         hostBuilder.Logging.ConfigureLogging(LogLevel.Information);

src/ServiceControl.Persistence/InternalCustomCheckClassification.cs

@@ -43,6 +43,7 @@ public static class InternalCustomCheckClassification
                 "Error Database Search Engine",  // RavenDB persister
                 "ServiceControl body storage",   // EF Core persisters
                 "Dead Letter Queue",             // ASBS / IBMMQ / MSMQ
+                "ServiceControl Retention",      // EF Core persisters
 
                 // ----- Audit instance (forwarded to the primary via ReportCustomCheckResult) -----
                 "Audit Message Ingestion",

src/ServiceControl.UnitTests/CustomChecks/InternalCustomCheckClassificationTests.cs

@@ -33,6 +33,7 @@ public void Internal_checks_are_flagged_internal()
         [TestCase("Error Message Ingestion")]
         [TestCase("Dead Letter Queue")]
         [TestCase("ServiceControl body storage")]
+        [TestCase("ServiceControl Retention")]
         [TestCase("Audit Message Ingestion Process")]
         public void Every_shipped_check_is_internal(string id)
         {

src/ServiceControl/App.config

@@ -23,7 +23,9 @@ These settings are only here so that we can debug ServiceControl while developin
     <!--<add key="ServiceControl/TransportType" value="RabbitMQ.QuorumConventionalRouting" />-->
     <!--<add key="ServiceControl/TransportType" value="SQLServer" />-->
 
-    <add key="ServiceControl/PersistenceType" value="RavenDB" />
+    <add key="ServiceControl/PersistenceType" value="RavenDB" /> 
+    <!--<add key="ServiceControl/PersistenceType" value="SQLServer" />-->
+    <!--<add key="ServiceControl/PersistenceType" value="PostgreSQL" />-->
 
     <!-- options are any comma separated combination of NLog,Seq,Otlp -->
     <add key="ServiceControl/LoggingProviders" value="NLog,Seq"/>
@@ -84,6 +86,8 @@ These settings are only here so that we can debug ServiceControl while developin
     <!-- Learning -->
     <!-- If the LearningTransport connectionString is empty, it will default to the solution directory when running/debugging from the IDE -->
     <add name="NServiceBus/Transport" connectionString="" />
+
+
 
     <!--Amazon SQS -->
     <!--<add name="NServiceBus/Transport" connectionString="Region=;QueueNamePrefix=;TopicNamePrefix=;AccessKeyId=;SecretAccessKey=;S3BucketForLargeMessages=;S3KeyPrefix=" />-->