Load Balancing: Implement distributed background jobs

src/Umbraco.Cms.Api.Common/DependencyInjection/UmbracoBuilderAuthExtensions.cs

@@ -9,8 +9,8 @@
 using Umbraco.Cms.Core;
 using Umbraco.Cms.Core.Configuration.Models;
 using Umbraco.Cms.Core.DependencyInjection;
+using Umbraco.Cms.Infrastructure.BackgroundJobs;
 using Umbraco.Cms.Infrastructure.BackgroundJobs.Jobs;
-using Umbraco.Extensions;
 
 namespace Umbraco.Cms.Api.Common.DependencyInjection;
 
@@ -139,7 +139,7 @@ private static void ConfigureOpenIddict(IUmbracoBuilder builder)
                 });
             });
 
-        builder.Services.AddRecurringBackgroundJob<OpenIddictCleanupJob>();
+        builder.Services.AddSingleton<IDistributedBackgroundJob, OpenIddictCleanupJob>();
         builder.Services.ConfigureOptions<ConfigureOpenIddict>();
     }
 }

src/Umbraco.Core/Configuration/Models/DistributedJobSettings.cs

@@ -0,0 +1,25 @@
+using System.ComponentModel;
+
+namespace Umbraco.Cms.Core.Configuration.Models;
+
+/// <summary>
+/// Settings for distributed jobs.
+/// </summary>
+[UmbracoOptions(Constants.Configuration.ConfigDistributedJobs)]
+public class DistributedJobSettings
+{
+    internal const string StaticPeriod = "00:00:10";
+    internal const string StaticDelay = "00:01:00";
+
+    /// <summary>
+    ///     Gets or sets a value for the period of checking if there are any runnable distributed jobs.
+    /// </summary>
+    [DefaultValue(StaticPeriod)]
+    public TimeSpan Period { get; set; } = TimeSpan.Parse(StaticPeriod);
+
+    /// <summary>
+    ///     Gets or sets a value for the delay of when to start checking for distributed jobs.
+    /// </summary>
+    [DefaultValue(StaticDelay)]
+    public TimeSpan Delay { get; set; } = TimeSpan.Parse(StaticDelay);
+}

src/Umbraco.Core/Constants-Configuration.cs

@@ -65,6 +65,7 @@ public static class Configuration
         public const string ConfigWebhook = ConfigPrefix + "Webhook";
         public const string ConfigWebhookPayloadType = ConfigWebhook + ":PayloadType";
         public const string ConfigCache = ConfigPrefix + "Cache";
+        public const string ConfigDistributedJobs = ConfigPrefix + "DistributedJobs";
 
         public static class NamedOptions
         {

src/Umbraco.Core/DependencyInjection/UmbracoBuilder.Configuration.cs

@@ -87,7 +87,8 @@ public static IUmbracoBuilder AddConfiguration(this IUmbracoBuilder builder)
             .AddUmbracoOptions<DataTypesSettings>()
             .AddUmbracoOptions<WebhookSettings>()
             .AddUmbracoOptions<CacheSettings>()
-            .AddUmbracoOptions<SystemDateMigrationSettings>();
+            .AddUmbracoOptions<SystemDateMigrationSettings>()
+            .AddUmbracoOptions<DistributedJobSettings>();
 
         // Configure connection string and ensure it's updated when the configuration changes
         builder.Services.AddSingleton<IConfigureOptions<ConnectionStrings>, ConfigureConnectionStrings>();

src/Umbraco.Core/Persistence/Constants-DatabaseSchema.cs

@@ -103,8 +103,8 @@ public static class Tables
             public const string Webhook2Headers = Webhook + "2Headers";
             public const string WebhookLog = Webhook + "Log";
             public const string WebhookRequest = Webhook + "Request";
-
             public const string LongRunningOperation = TableNamePrefix + "LongRunningOperation";
+            public const string DistributedJob = TableNamePrefix + "DistributedJob";
         }
     }
 }

src/Umbraco.Core/Persistence/Constants-Locks.cs

@@ -90,5 +90,10 @@ public static class Locks
         ///     All document URLs.
         /// </summary>
         public const int DocumentUrls = -345;
+
+        /// <summary>
+        ///     All distributed jobs.
+        /// </summary>
+        public const int DistributedJobs = -346;
     }
 }

src/Umbraco.Infrastructure/BackgroundJobs/DistributedBackgroundJobHostedService.cs

@@ -0,0 +1,103 @@
+using System.Diagnostics;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Umbraco.Cms.Core;
+using Umbraco.Cms.Core.Configuration.Models;
+using Umbraco.Cms.Core.Services;
+using Umbraco.Cms.Infrastructure.Services;
+
+namespace Umbraco.Cms.Infrastructure.BackgroundJobs;
+
+/// <summary>
+///    A hosted service that checks for any runnable distributed background jobs on a timer.
+/// </summary>
+public class DistributedBackgroundJobHostedService : BackgroundService
+{
+    private readonly ILogger<DistributedBackgroundJobHostedService> _logger;
+    private readonly IRuntimeState _runtimeState;
+    private readonly IDistributedJobService _distributedJobService;
+    private DistributedJobSettings _distributedJobSettings;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="DistributedBackgroundJobHostedService"/> class.
+    /// </summary>
+    /// <param name="logger"></param>
+    /// <param name="runtimeState"></param>
+    /// <param name="distributedJobService"></param>
+    /// <param name="distributedJobSettings"></param>
+    public DistributedBackgroundJobHostedService(
+        ILogger<DistributedBackgroundJobHostedService> logger,
+        IRuntimeState runtimeState,
+        IDistributedJobService distributedJobService,
+        IOptionsMonitor<DistributedJobSettings> distributedJobSettings)
+    {
+        _logger = logger;
+        _runtimeState = runtimeState;
+        _distributedJobService = distributedJobService;
+        _distributedJobSettings = distributedJobSettings.CurrentValue;
+        distributedJobSettings.OnChange(options =>
+        {
+            _distributedJobSettings = options;
+        });
+    }
+
+    /// <inheritdoc />
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await Task.Delay(_distributedJobSettings.Delay, stoppingToken);
+        // Update all jobs, periods might have changed when restarting.
+        await _distributedJobService.UpdateAllChangedAsync();
+
+        using PeriodicTimer timer = new(_distributedJobSettings.Period);
+
+        try
+        {
+            while (await timer.WaitForNextTickAsync(stoppingToken))
+            {
+                await RunRunnableJob();
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            _logger.LogInformation("Timed Hosted Service is stopping.");
+        }
+    }
+
+    private async Task RunRunnableJob()
+    {
+        // Do not run distributed jobs if we aren't in Run level, as we might not have booted properly.
+        if (_runtimeState.Level != RuntimeLevel.Run)
+        {
+            return;
+        }
+
+        IDistributedBackgroundJob? job = await _distributedJobService.TryTakeRunnableAsync();
+
+        if (job is null)
+        {
+            // No runnable jobs for now, return
+            return;
+        }
+
+        try
+        {
+            await job.RunJobAsync();
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "An exception occurred while running distributed background job '{JobName}'.", job.Name);
+        }
+        finally
+        {
+            try
+            {
+                await _distributedJobService.FinishAsync(job.Name);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "An exception occurred while finishing distributed background job '{JobName}'.", job.Name);
+            }
+        }
+    }
+}

src/Umbraco.Infrastructure/BackgroundJobs/IDistributedBackgroundJob.cs

@@ -0,0 +1,24 @@
+namespace Umbraco.Cms.Infrastructure.BackgroundJobs;
+
+/// <summary>
+/// A background job that will be executed by an available server. With a single server setup this will always be the same.
+/// With a load balanced setup, the executing server might change every time this needs to be executed.
+/// </summary>
+public interface IDistributedBackgroundJob
+{
+    /// <summary>
+    /// Name of the job
+    /// </summary>
+    string Name { get; }
+
+    /// <summary>
+    /// Timespan representing how often the task should recur.
+    /// </summary>
+    TimeSpan Period { get; }
+
+    /// <summary>
+    /// Run the job.
+    /// </summary>
+    /// <returns></returns>
+    Task RunJobAsync();
+}

src/Umbraco.Infrastructure/BackgroundJobs/Jobs/CacheInstructionsPruningJob.cs

@@ -2,14 +2,13 @@
 using Umbraco.Cms.Core.Configuration.Models;
 using Umbraco.Cms.Core.Persistence.Repositories;
 using Umbraco.Cms.Core.Scoping;
-using Umbraco.Cms.Infrastructure.Scoping;
 
 namespace Umbraco.Cms.Infrastructure.BackgroundJobs.Jobs;
 
 /// <summary>
 /// A background job that prunes cache instructions from the database.
 /// </summary>
-public class CacheInstructionsPruningJob : IRecurringBackgroundJob
+public class CacheInstructionsPruningJob : IDistributedBackgroundJob
 {
     private readonly IOptions<GlobalSettings> _globalSettings;
     private readonly ICacheInstructionRepository _cacheInstructionRepository;
@@ -36,12 +35,7 @@ public CacheInstructionsPruningJob(
         Period = globalSettings.Value.DatabaseServerMessenger.TimeBetweenPruneOperations;
     }
 
-    /// <inheritdoc />
-    public event EventHandler PeriodChanged
-    {
-        add { }
-        remove { }
-    }
+    public string Name => "CacheInstructionsPruningJob";
 
     /// <inheritdoc />
     public TimeSpan Period { get; }

src/Umbraco.Infrastructure/BackgroundJobs/Jobs/ContentVersionCleanupJob.cs

@@ -1,24 +1,21 @@
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using Umbraco.Cms.Core;
 using Umbraco.Cms.Core.Configuration.Models;
-using Umbraco.Cms.Core.Runtime;
 using Umbraco.Cms.Core.Services;
-using Umbraco.Cms.Core.Sync;
-using Umbraco.Cms.Infrastructure.BackgroundJobs;
 
 namespace Umbraco.Cms.Infrastructure.BackgroundJobs.Jobs;
 
 /// <summary>
 ///     Recurring hosted service that executes the content history cleanup.
 /// </summary>
-public class ContentVersionCleanupJob : IRecurringBackgroundJob
+public class ContentVersionCleanupJob : IDistributedBackgroundJob
 {
+    /// <inheritdoc />
+    public string Name => "ContentVersionCleanupJob";
 
+    /// <inheritdoc />
     public TimeSpan Period { get => TimeSpan.FromHours(1); }
 
-    // No-op event as the period never changes on this job
-    public event EventHandler PeriodChanged { add { } remove { } }
 
     private readonly ILogger<ContentVersionCleanupJob> _logger;
     private readonly IContentVersionService _service;

src/Umbraco.Infrastructure/BackgroundJobs/Jobs/HealthCheckNotifierJob.cs

@@ -17,17 +17,11 @@ namespace Umbraco.Cms.Infrastructure.BackgroundJobs.Jobs;
 /// <summary>
 ///     Hosted service implementation for recurring health check notifications.
 /// </summary>
-public class HealthCheckNotifierJob : IRecurringBackgroundJob
+public class HealthCheckNotifierJob : IDistributedBackgroundJob
 {
+    /// <inheritdoc />
+    public string Name => "HealthCheckNotifierJob";
     public TimeSpan Period { get; private set; }
-    public TimeSpan Delay { get; private set; }
-
-    private event EventHandler? _periodChanged;
-    public event EventHandler PeriodChanged
-    {
-        add { _periodChanged += value; }
-        remove { _periodChanged -= value; }
-    }
 
     private readonly HealthCheckCollection _healthChecks;
     private readonly ILogger<HealthCheckNotifierJob> _logger;
@@ -67,14 +61,12 @@ public HealthCheckNotifierJob(
         _eventAggregator = eventAggregator;
 
         Period = healthChecksSettings.CurrentValue.Notification.Period;
-        Delay = DelayCalculator.GetDelay(healthChecksSettings.CurrentValue.Notification.FirstRunTime, cronTabParser, logger, TimeSpan.FromMinutes(3));
 
 
         healthChecksSettings.OnChange(x =>
         {
             _healthChecksSettings = x;
             Period = x.Notification.Period;
-            _periodChanged?.Invoke(this, EventArgs.Empty);
         });
     }
 

src/Umbraco.Infrastructure/BackgroundJobs/Jobs/LogScrubberJob.cs

@@ -3,13 +3,10 @@
 
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using Umbraco.Cms.Core;
 using Umbraco.Cms.Core.Configuration.Models;
 using Umbraco.Cms.Core.Logging;
-using Umbraco.Cms.Core.Runtime;
 using Umbraco.Cms.Core.Scoping;
 using Umbraco.Cms.Core.Services;
-using Umbraco.Cms.Core.Sync;
 
 namespace Umbraco.Cms.Infrastructure.BackgroundJobs.Jobs;
 
@@ -19,18 +16,18 @@ namespace Umbraco.Cms.Infrastructure.BackgroundJobs.Jobs;
 /// <remarks>
 ///     Will only run on non-replica servers.
 /// </remarks>
-public class LogScrubberJob : IRecurringBackgroundJob
+public class LogScrubberJob : IDistributedBackgroundJob
 {
     private readonly IAuditService _auditService;
     private readonly ILogger<LogScrubberJob> _logger;
     private readonly IProfilingLogger _profilingLogger;
     private readonly ICoreScopeProvider _scopeProvider;
     private LoggingSettings _settings;
 
-    public TimeSpan Period => TimeSpan.FromHours(4);
+    /// <inheritdoc />
+    public string Name => "LogScrubberJob";
 
-    // No-op event as the period never changes on this job
-    public event EventHandler PeriodChanged { add { } remove { } }
+    public TimeSpan Period => TimeSpan.FromHours(4);
 
     /// <summary>
     ///     Initializes a new instance of the <see cref="LogScrubberJob" /> class.

src/Umbraco.Infrastructure/BackgroundJobs/Jobs/LongRunningOperationsCleanupJob.cs

@@ -8,7 +8,7 @@ namespace Umbraco.Cms.Infrastructure.BackgroundJobs.Jobs;
 /// <summary>
 /// Cleans up long-running operations that have exceeded a specified age.
 /// </summary>
-public class LongRunningOperationsCleanupJob : IRecurringBackgroundJob
+public class LongRunningOperationsCleanupJob : IDistributedBackgroundJob
 {
     private readonly ICoreScopeProvider _scopeProvider;
     private readonly ILongRunningOperationRepository _longRunningOperationRepository;
@@ -36,18 +36,11 @@ public LongRunningOperationsCleanupJob(
     }
 
     /// <inheritdoc />
-    public event EventHandler? PeriodChanged
-    {
-        add { }
-        remove { }
-    }
+    public string Name => "LongRunningOperationsCleanupJob";
 
     /// <inheritdoc />
     public TimeSpan Period { get; }
 
-    /// <inheritdoc/>
-    public TimeSpan Delay { get; } = TimeSpan.FromSeconds(10);
-
     /// <inheritdoc />
     public async Task RunJobAsync()
     {