feat: gate periodic library tasks behind a scan-leader lease

Timer-driven library tasks fire on every replica, so a library refresh or a
database optimise runs once per pod against the same shared library.

- add IScanLeaderLease with a Redis TTL implementation and a no-op default
- skip timer-driven runs of the gated tasks on instances without the lease
- treat an unreachable Redis as holding the lease so tasks never stop running
- leave manual and API-triggered runs ungated
This commit is contained in:
2026-09-11 23:56:51 +10:00
parent e3571eb2a4
commit 3b416c7fb2
10 changed files with 706 additions and 3 deletions
@@ -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;
}
}
}
@@ -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);
@@ -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();
}