unkin-agent
3b416c7fb2
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
82 lines
3.1 KiB
C#
82 lines
3.1 KiB
C#
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;
|
|
}
|
|
}
|
|
}
|