Gate periodic library-mutating tasks behind a scan-leader lease

In a multi-pod deployment every pod runs the scheduled-task timers, so
periodic library-mutating tasks (library refresh, people/chapter refresh,
audio normalization, media-segment and keyframe extraction, collection and
user-data cleanup, database optimization) fire concurrently against the shared
database and library, duplicating work and racing each other.

Add an IScanLeaderLease abstraction that elects a single scan leader via a
Redis TTL lease keyed on the pod identity, mirroring the existing transcode
lease machinery. RedisScanLeaderLease acquires or renews the lease with an
atomic Lua script and fails safe by treating the pod as leader whenever Redis
is unreachable, so scans never stall. NullScanLeaderLease preserves the
single-instance behavior when election is disabled or no Redis connection is
configured.

Gate only the timer-driven path in ScheduledTaskWorker: when election is
enabled and a task key is in the gated set, a non-leader re-arms its trigger
and skips enqueueing. Manual and API-triggered runs bypass this path and still
run on any pod. Wiring is additive and the new worker constructor parameters
are optional, so existing behavior is unchanged when election is off.

Signed-off-by: Ben Vincent <ben@unkin.net>

Emby.Server.Implementations/ScheduledTasks/RedisScanLeaderLease.cs

@@ -0,0 +1,81 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Controller.ScheduledTasks;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using StackExchange.Redis;
+
+namespace Emby.Server.Implementations.ScheduledTasks;
+
+/// <summary>
+/// A Redis-backed <see cref="IScanLeaderLease"/> that elects a single scan-leader instance using a
+/// TTL lease on a shared key. The lease value is this instance's pod identity; a leader that keeps
+/// renewing retains the lease, and any instance can claim it once the previous leader's lease expires.
+/// </summary>
+public sealed class RedisScanLeaderLease : IScanLeaderLease
+{
+    private const string LeaderKey = "jellyfin:scanleader";
+
+    /// <summary>
+    /// Lua script for atomic acquire-or-renew: if the key is unset (missing or already expired) it is
+    /// set to this pod for the lease duration and 1 is returned; if it already holds this pod the TTL is
+    /// extended and 1 is returned; otherwise another pod owns a live lease and 0 is returned.
+    /// </summary>
+    private const string AcquireOrRenewScript = @"
+local current = redis.call('GET', KEYS[1])
+if not current then
+    redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])
+    return 1
+elseif current == ARGV[1] then
+    redis.call('PEXPIRE', KEYS[1], ARGV[2])
+    return 1
+else
+    return 0
+end";
+
+    private readonly IDatabase _db;
+    private readonly ScanLeaderOptions _options;
+    private readonly ILogger<RedisScanLeaderLease> _logger;
+    private readonly string _podId;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RedisScanLeaderLease"/> class.
+    /// </summary>
+    /// <param name="redis">The Redis connection multiplexer.</param>
+    /// <param name="options">The scan-leader configuration options.</param>
+    /// <param name="logger">The logger.</param>
+    public RedisScanLeaderLease(
+        IConnectionMultiplexer redis,
+        IOptions<ScanLeaderOptions> options,
+        ILogger<RedisScanLeaderLease> logger)
+    {
+        _db = redis.GetDatabase();
+        _options = options.Value;
+        _logger = logger;
+        _podId = Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") ?? Environment.MachineName;
+    }
+
+    /// <inheritdoc />
+    public async Task<bool> TryAcquireOrRenewAsync(CancellationToken cancellationToken = default)
+    {
+        var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
+
+        try
+        {
+            var result = (long?)await _db.ScriptEvaluateAsync(
+                AcquireOrRenewScript,
+                keys: new RedisKey[] { LeaderKey },
+                values: new RedisValue[] { _podId, leaseDurationMs }).ConfigureAwait(false);
+
+            return result == 1;
+        }
+        catch (Exception ex)
+        {
+            // Fail-safe: if Redis is unreachable, treat this instance as the leader so scheduled scans
+            // keep running. Every instance scanning is preferable to no instance scanning.
+            _logger.LogWarning(ex, "Scan-leader lease evaluation failed; treating {PodId} as leader.", _podId);
+            return true;
+        }
+    }
+}

Emby.Server.Implementations/ScheduledTasks/ScheduledTaskWorker.cs

@@ -13,6 +13,7 @@ using Jellyfin.Data.Events;
 using Jellyfin.Extensions.Json;
 using MediaBrowser.Common.Configuration;
 using MediaBrowser.Common.Extensions;
+using MediaBrowser.Controller.ScheduledTasks;
 using MediaBrowser.Model.Tasks;
 using Microsoft.Extensions.Logging;
 
@@ -27,6 +28,8 @@ public class ScheduledTaskWorker : IScheduledTaskWorker
     private readonly IApplicationPaths _applicationPaths;
     private readonly ILogger _logger;
     private readonly ITaskManager _taskManager;
+    private readonly IScanLeaderLease _scanLeaderLease;
+    private readonly ScanLeaderOptions _scanLeaderOptions;
     private readonly Lock _lastExecutionResultSyncLock = new();
     private bool _readFromFile;
     private TaskResult _lastExecutionResult;
@@ -41,6 +44,8 @@ public class ScheduledTaskWorker : IScheduledTaskWorker
     /// <param name="applicationPaths">The application paths.</param>
     /// <param name="taskManager">The task manager.</param>
     /// <param name="logger">The logger.</param>
+    /// <param name="scanLeaderLease">The scan-leader lease used to gate periodic library-mutating tasks, or <c>null</c> to disable gating.</param>
+    /// <param name="scanLeaderOptions">The scan-leader options, or <c>null</c> to disable gating.</param>
     /// <exception cref="ArgumentNullException">
     /// scheduledTask
     /// or
@@ -52,7 +57,13 @@ public class ScheduledTaskWorker : IScheduledTaskWorker
     /// or
     /// logger.
     /// </exception>
-    public ScheduledTaskWorker(IScheduledTask scheduledTask, IApplicationPaths applicationPaths, ITaskManager taskManager, ILogger logger)
+    public ScheduledTaskWorker(
+        IScheduledTask scheduledTask,
+        IApplicationPaths applicationPaths,
+        ITaskManager taskManager,
+        ILogger logger,
+        IScanLeaderLease scanLeaderLease = null,
+        ScanLeaderOptions scanLeaderOptions = null)
     {
         ArgumentNullException.ThrowIfNull(scheduledTask);
         ArgumentNullException.ThrowIfNull(applicationPaths);
@@ -63,6 +74,8 @@ public class ScheduledTaskWorker : IScheduledTaskWorker
         _applicationPaths = applicationPaths;
         _taskManager = taskManager;
         _logger = logger;
+        _scanLeaderLease = scanLeaderLease;
+        _scanLeaderOptions = scanLeaderOptions;
 
         InitTriggerEvents();
     }
@@ -268,6 +281,20 @@ public class ScheduledTaskWorker : IScheduledTaskWorker
 
         trigger.Stop();
 
+        if (_scanLeaderLease is not null
+            && _scanLeaderOptions is not null
+            && _scanLeaderOptions.Enabled
+            && _scanLeaderOptions.GatedTaskKeys is not null
+            && _scanLeaderOptions.GatedTaskKeys.Contains(ScheduledTask.Key, StringComparer.Ordinal)
+            && !await _scanLeaderLease.TryAcquireOrRenewAsync().ConfigureAwait(false))
+        {
+            _logger.LogDebug("Skipping gated task {Task}: this instance does not hold the scan-leader lease.", Name);
+
+            // Re-arm the trigger for the next interval without enqueueing on this instance.
+            trigger.Start(LastExecutionResult, _logger, Name, false);
+            return;
+        }
+
         _taskManager.QueueScheduledTask(ScheduledTask, trigger.TaskOptions);
 
         await Task.Delay(1000).ConfigureAwait(false);

Emby.Server.Implementations/ScheduledTasks/TaskManager.cs

@@ -5,8 +5,10 @@ using System.Linq;
 using System.Threading.Tasks;
 using Jellyfin.Data.Events;
 using MediaBrowser.Common.Configuration;
+using MediaBrowser.Controller.ScheduledTasks;
 using MediaBrowser.Model.Tasks;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 
 namespace Emby.Server.Implementations.ScheduledTasks;
 
@@ -23,18 +25,26 @@ public class TaskManager : ITaskManager
 
     private readonly IApplicationPaths _applicationPaths;
     private readonly ILogger<TaskManager> _logger;
+    private readonly IScanLeaderLease? _scanLeaderLease;
+    private readonly ScanLeaderOptions? _scanLeaderOptions;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="TaskManager" /> class.
     /// </summary>
     /// <param name="applicationPaths">The application paths.</param>
     /// <param name="logger">The logger.</param>
+    /// <param name="scanLeaderLease">The scan-leader lease used to gate periodic library-mutating tasks, or <c>null</c> to disable gating.</param>
+    /// <param name="scanLeaderOptions">The scan-leader options, or <c>null</c> to disable gating.</param>
     public TaskManager(
         IApplicationPaths applicationPaths,
-        ILogger<TaskManager> logger)
+        ILogger<TaskManager> logger,
+        IScanLeaderLease? scanLeaderLease = null,
+        IOptions<ScanLeaderOptions>? scanLeaderOptions = null)
     {
         _applicationPaths = applicationPaths;
         _logger = logger;
+        _scanLeaderLease = scanLeaderLease;
+        _scanLeaderOptions = scanLeaderOptions?.Value;
 
         ScheduledTasks = [];
     }
@@ -175,7 +185,7 @@ public class TaskManager : ITaskManager
     /// <inheritdoc />
     public void AddTasks(IEnumerable<IScheduledTask> tasks)
     {
-        var list = tasks.Select(t => new ScheduledTaskWorker(t, _applicationPaths, this, _logger));
+        var list = tasks.Select(t => new ScheduledTaskWorker(t, _applicationPaths, this, _logger, _scanLeaderLease, _scanLeaderOptions));
 
         ScheduledTasks = ScheduledTasks.Concat(list).ToArray();
     }