From 0008bde28e41bcfb42c6de8dcb335c7a78ec3241 Mon Sep 17 00:00:00 2001 From: Ben Vincent Date: Mon, 10 Aug 2026 23:51:18 +1000 Subject: [PATCH] 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 --- .../ScheduledTasks/RedisScanLeaderLease.cs | 81 ++++++ .../ScheduledTasks/ScheduledTaskWorker.cs | 29 +- .../ScheduledTasks/TaskManager.cs | 14 +- Jellyfin.Server/CoreAppHost.cs | 15 + .../ScheduledTasks/IScanLeaderLease.cs | 21 ++ .../ScheduledTasks/NullScanLeaderLease.cs | 16 ++ .../ScheduledTasks/ScanLeaderOptions.cs | 38 +++ .../ScheduledTasks/ScanLeaderLeaseTests.cs | 256 ++++++++++++++++++ .../ScheduledTaskWorkerLeaderGatingTests.cs | 189 +++++++++++++ 9 files changed, 656 insertions(+), 3 deletions(-) create mode 100644 Emby.Server.Implementations/ScheduledTasks/RedisScanLeaderLease.cs create mode 100644 MediaBrowser.Controller/ScheduledTasks/IScanLeaderLease.cs create mode 100644 MediaBrowser.Controller/ScheduledTasks/NullScanLeaderLease.cs create mode 100644 MediaBrowser.Controller/ScheduledTasks/ScanLeaderOptions.cs create mode 100644 tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScanLeaderLeaseTests.cs create mode 100644 tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScheduledTaskWorkerLeaderGatingTests.cs diff --git a/Emby.Server.Implementations/ScheduledTasks/RedisScanLeaderLease.cs b/Emby.Server.Implementations/ScheduledTasks/RedisScanLeaderLease.cs new file mode 100644 index 000000000..a95e06fe5 --- /dev/null +++ b/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; + +/// +/// A Redis-backed 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. +/// +public sealed class RedisScanLeaderLease : IScanLeaderLease +{ + private const string LeaderKey = "jellyfin:scanleader"; + + /// + /// 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. + /// + 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 _logger; + private readonly string _podId; + + /// + /// Initializes a new instance of the class. + /// + /// The Redis connection multiplexer. + /// The scan-leader configuration options. + /// The logger. + public RedisScanLeaderLease( + IConnectionMultiplexer redis, + IOptions options, + ILogger logger) + { + _db = redis.GetDatabase(); + _options = options.Value; + _logger = logger; + _podId = Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") ?? Environment.MachineName; + } + + /// + public async Task 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; + } + } +} diff --git a/Emby.Server.Implementations/ScheduledTasks/ScheduledTaskWorker.cs b/Emby.Server.Implementations/ScheduledTasks/ScheduledTaskWorker.cs index 24f554981..ade28349a 100644 --- a/Emby.Server.Implementations/ScheduledTasks/ScheduledTaskWorker.cs +++ b/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 /// The application paths. /// The task manager. /// The logger. + /// The scan-leader lease used to gate periodic library-mutating tasks, or null to disable gating. + /// The scan-leader options, or null to disable gating. /// /// scheduledTask /// or @@ -52,7 +57,13 @@ public class ScheduledTaskWorker : IScheduledTaskWorker /// or /// logger. /// - 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); diff --git a/Emby.Server.Implementations/ScheduledTasks/TaskManager.cs b/Emby.Server.Implementations/ScheduledTasks/TaskManager.cs index 4ec2c9c78..d7bb9a1f9 100644 --- a/Emby.Server.Implementations/ScheduledTasks/TaskManager.cs +++ b/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 _logger; + private readonly IScanLeaderLease? _scanLeaderLease; + private readonly ScanLeaderOptions? _scanLeaderOptions; /// /// Initializes a new instance of the class. /// /// The application paths. /// The logger. + /// The scan-leader lease used to gate periodic library-mutating tasks, or null to disable gating. + /// The scan-leader options, or null to disable gating. public TaskManager( IApplicationPaths applicationPaths, - ILogger logger) + ILogger logger, + IScanLeaderLease? scanLeaderLease = null, + IOptions? scanLeaderOptions = null) { _applicationPaths = applicationPaths; _logger = logger; + _scanLeaderLease = scanLeaderLease; + _scanLeaderOptions = scanLeaderOptions?.Value; ScheduledTasks = []; } @@ -175,7 +185,7 @@ public class TaskManager : ITaskManager /// public void AddTasks(IEnumerable 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(); } diff --git a/Jellyfin.Server/CoreAppHost.cs b/Jellyfin.Server/CoreAppHost.cs index 6020a1bc3..0077b52bf 100644 --- a/Jellyfin.Server/CoreAppHost.cs +++ b/Jellyfin.Server/CoreAppHost.cs @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.Reflection; using Emby.Server.Implementations; using Emby.Server.Implementations.MediaEncoding; +using Emby.Server.Implementations.ScheduledTasks; using Emby.Server.Implementations.Session; using Jellyfin.Api.WebSocketListeners; using Jellyfin.Database.Implementations; @@ -26,6 +27,7 @@ using MediaBrowser.Controller.Library; using MediaBrowser.Controller.Lyrics; using MediaBrowser.Controller.MediaEncoding; using MediaBrowser.Controller.Net; +using MediaBrowser.Controller.ScheduledTasks; using MediaBrowser.Controller.Security; using MediaBrowser.Controller.Trickplay; using MediaBrowser.Model.Activity; @@ -129,6 +131,19 @@ namespace Jellyfin.Server serviceCollection.AddSingleton(); } + // Scan-leader lease: gates periodic library-mutating scheduled tasks to a single leader + // instance. Redis-backed when enabled and a Redis connection is configured, no-op otherwise. + serviceCollection.Configure(_startupConfig.GetSection("Jellyfin:ScanLeader")); + var scanLeaderEnabled = bool.TryParse(_startupConfig["Jellyfin:ScanLeader:Enabled"], out var enabled) && enabled; + if (scanLeaderEnabled && !string.IsNullOrEmpty(redisConnectionString)) + { + serviceCollection.AddSingleton(); + } + else + { + serviceCollection.AddSingleton(); + } + foreach (var type in GetExportTypes()) { serviceCollection.AddSingleton(typeof(ILyricProvider), type); diff --git a/MediaBrowser.Controller/ScheduledTasks/IScanLeaderLease.cs b/MediaBrowser.Controller/ScheduledTasks/IScanLeaderLease.cs new file mode 100644 index 000000000..bbd71ef13 --- /dev/null +++ b/MediaBrowser.Controller/ScheduledTasks/IScanLeaderLease.cs @@ -0,0 +1,21 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace MediaBrowser.Controller.ScheduledTasks; + +/// +/// Provides a distributed leader lease that gates periodic, library-mutating scheduled tasks +/// to a single instance across a multi-pod deployment. +/// +public interface IScanLeaderLease +{ + /// + /// Attempts to acquire the scan-leader lease, or renews it when this instance already holds it. + /// + /// A cancellation token. + /// + /// true if this instance holds the leader lease and gated periodic tasks may run here; + /// otherwise false. + /// + Task TryAcquireOrRenewAsync(CancellationToken cancellationToken = default); +} diff --git a/MediaBrowser.Controller/ScheduledTasks/NullScanLeaderLease.cs b/MediaBrowser.Controller/ScheduledTasks/NullScanLeaderLease.cs new file mode 100644 index 000000000..04eb662fc --- /dev/null +++ b/MediaBrowser.Controller/ScheduledTasks/NullScanLeaderLease.cs @@ -0,0 +1,16 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace MediaBrowser.Controller.ScheduledTasks; + +/// +/// A no-op used when scan-leader election is disabled or no Redis +/// connection is configured. Every instance is treated as the leader, preserving the default +/// single-instance behavior where all periodic tasks run locally. +/// +public sealed class NullScanLeaderLease : IScanLeaderLease +{ + /// + public Task TryAcquireOrRenewAsync(CancellationToken cancellationToken = default) + => Task.FromResult(true); +} diff --git a/MediaBrowser.Controller/ScheduledTasks/ScanLeaderOptions.cs b/MediaBrowser.Controller/ScheduledTasks/ScanLeaderOptions.cs new file mode 100644 index 000000000..626601fa7 --- /dev/null +++ b/MediaBrowser.Controller/ScheduledTasks/ScanLeaderOptions.cs @@ -0,0 +1,38 @@ +namespace MediaBrowser.Controller.ScheduledTasks; + +/// +/// Configuration options for scan-leader election, which gates periodic library-mutating +/// scheduled tasks to a single leader instance in a multi-pod deployment. +/// +public sealed class ScanLeaderOptions +{ + /// + /// Gets or sets a value indicating whether scan-leader election is enabled. When disabled, + /// every instance runs its periodic tasks as before. + /// + public bool Enabled { get; set; } + + /// + /// Gets or sets the duration in seconds for which the scan-leader lease is held before it must + /// be renewed. A leader that stops renewing loses the lease after this duration. + /// + public int LeaseDurationSeconds { get; set; } = 60; + + /// + /// Gets or sets the set of scheduled task keys whose periodic (timer-driven) execution is gated + /// to the scan leader. Tasks not listed here run on every instance, and manual or API-triggered + /// runs are never gated. + /// + public string[] GatedTaskKeys { get; set; } = + { + "RefreshLibrary", + "RefreshPeople", + "RefreshChapterImages", + "AudioNormalization", + "TaskExtractMediaSegments", + "KeyframeExtraction", + "CleanCollectionsAndPlaylists", + "CleanupUserDataTask", + "OptimizeDatabaseTask" + }; +} diff --git a/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScanLeaderLeaseTests.cs b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScanLeaderLeaseTests.cs new file mode 100644 index 000000000..3ee444a6d --- /dev/null +++ b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScanLeaderLeaseTests.cs @@ -0,0 +1,256 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Emby.Server.Implementations.ScheduledTasks; +using MediaBrowser.Controller.ScheduledTasks; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Moq; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Implementations.Tests.ScheduledTasks; + +/// +/// Tests for scan-leader lease behavior. The acquire/renew/takeover state machine is exercised +/// through an in-memory reference implementation that mirrors the Redis Lua contract (no real Redis +/// required), while the fail-safe and success paths of are +/// exercised against a mocked . +/// +public class ScanLeaderLeaseTests +{ + /// + /// Verifies that the first instance to call the lease becomes the leader. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task Acquire_WhenUnheld_ReturnsTrue() + { + var store = new FakeLeaderStore(); + var clock = new TestClock(DateTime.UtcNow); + var podA = new ReferenceScanLeaderLease(store, "pod-a", clock, TimeSpan.FromSeconds(60)); + + Assert.True(await podA.TryAcquireOrRenewAsync()); + Assert.Equal("pod-a", store.Owner); + } + + /// + /// Verifies that the current leader renewing its own lease succeeds. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task Renew_BySameInstance_ReturnsTrue() + { + var store = new FakeLeaderStore(); + var clock = new TestClock(DateTime.UtcNow); + var podA = new ReferenceScanLeaderLease(store, "pod-a", clock, TimeSpan.FromSeconds(60)); + + Assert.True(await podA.TryAcquireOrRenewAsync()); + + clock.Advance(TimeSpan.FromSeconds(10)); + + Assert.True(await podA.TryAcquireOrRenewAsync()); + Assert.Equal("pod-a", store.Owner); + } + + /// + /// Verifies that a second instance cannot acquire the lease while the leader's lease is still valid. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task Acquire_BySecondInstance_WhileLeaseValid_ReturnsFalse() + { + var store = new FakeLeaderStore(); + var clock = new TestClock(DateTime.UtcNow); + var podA = new ReferenceScanLeaderLease(store, "pod-a", clock, TimeSpan.FromSeconds(60)); + var podB = new ReferenceScanLeaderLease(store, "pod-b", clock, TimeSpan.FromSeconds(60)); + + Assert.True(await podA.TryAcquireOrRenewAsync()); + + clock.Advance(TimeSpan.FromSeconds(30)); + + Assert.False(await podB.TryAcquireOrRenewAsync()); + Assert.Equal("pod-a", store.Owner); + } + + /// + /// Verifies that a second instance takes over the lease once the previous leader's lease has expired. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task Acquire_BySecondInstance_AfterLeaseExpires_ReturnsTrue() + { + var store = new FakeLeaderStore(); + var clock = new TestClock(DateTime.UtcNow); + var podA = new ReferenceScanLeaderLease(store, "pod-a", clock, TimeSpan.FromSeconds(60)); + var podB = new ReferenceScanLeaderLease(store, "pod-b", clock, TimeSpan.FromSeconds(60)); + + Assert.True(await podA.TryAcquireOrRenewAsync()); + + // Advance past pod-a's lease expiry without pod-a renewing. + clock.Advance(TimeSpan.FromSeconds(61)); + + Assert.True(await podB.TryAcquireOrRenewAsync()); + Assert.Equal("pod-b", store.Owner); + } + + /// + /// Verifies that always reports the caller as the leader. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task NullScanLeaderLease_AlwaysReturnsTrue() + { + var lease = new NullScanLeaderLease(); + + Assert.True(await lease.TryAcquireOrRenewAsync()); + Assert.True(await lease.TryAcquireOrRenewAsync()); + } + + /// + /// Verifies that returns true (fail-safe) when the Redis + /// evaluation throws, so that scheduled scans keep running when Redis is unreachable. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task RedisScanLeaderLease_WhenRedisThrows_ReturnsTrue() + { + var dbMock = new Mock(); + dbMock + .Setup(d => d.ScriptEvaluateAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .ThrowsAsync(new InvalidOperationException("Redis unavailable")); + + var lease = new RedisScanLeaderLease( + CreateMultiplexer(dbMock.Object), + Options.Create(new ScanLeaderOptions { Enabled = true, LeaseDurationSeconds = 60 }), + new Mock>().Object); + + Assert.True(await lease.TryAcquireOrRenewAsync()); + } + + /// + /// Verifies that reports leadership when the Redis script + /// returns 1 (lease acquired or renewed). + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task RedisScanLeaderLease_WhenScriptReturnsOne_ReturnsTrue() + { + var dbMock = new Mock(); + dbMock + .Setup(d => d.ScriptEvaluateAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .ReturnsAsync(RedisResult.Create((RedisValue)1L)); + + var lease = new RedisScanLeaderLease( + CreateMultiplexer(dbMock.Object), + Options.Create(new ScanLeaderOptions { Enabled = true, LeaseDurationSeconds = 60 }), + new Mock>().Object); + + Assert.True(await lease.TryAcquireOrRenewAsync()); + } + + /// + /// Verifies that reports non-leadership when the Redis script + /// returns 0 (another instance holds a live lease). + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task RedisScanLeaderLease_WhenScriptReturnsZero_ReturnsFalse() + { + var dbMock = new Mock(); + dbMock + .Setup(d => d.ScriptEvaluateAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .ReturnsAsync(RedisResult.Create((RedisValue)0L)); + + var lease = new RedisScanLeaderLease( + CreateMultiplexer(dbMock.Object), + Options.Create(new ScanLeaderOptions { Enabled = true, LeaseDurationSeconds = 60 }), + new Mock>().Object); + + Assert.False(await lease.TryAcquireOrRenewAsync()); + } + + private static IConnectionMultiplexer CreateMultiplexer(IDatabase database) + { + var muxMock = new Mock(); + muxMock + .Setup(m => m.GetDatabase(It.IsAny(), It.IsAny())) + .Returns(database); + return muxMock.Object; + } + + private sealed class FakeLeaderStore + { + public string? Owner { get; set; } + + public DateTime ExpiresUtc { get; set; } + } + + private sealed class TestClock + { + private DateTime _now; + + public TestClock(DateTime now) + { + _now = now; + } + + public DateTime UtcNow => _now; + + public void Advance(TimeSpan by) => _now += by; + } + + /// + /// In-memory reference lease that mirrors the Redis Lua acquire-or-renew contract: a key that is + /// unset or expired is claimed by the caller; a key already owned by the caller is renewed; a key + /// owned by a different, still-valid holder is refused. + /// + private sealed class ReferenceScanLeaderLease : IScanLeaderLease + { + private readonly FakeLeaderStore _store; + private readonly string _podId; + private readonly TestClock _clock; + private readonly TimeSpan _ttl; + + public ReferenceScanLeaderLease(FakeLeaderStore store, string podId, TestClock clock, TimeSpan ttl) + { + _store = store; + _podId = podId; + _clock = clock; + _ttl = ttl; + } + + public Task TryAcquireOrRenewAsync(CancellationToken cancellationToken = default) + { + var now = _clock.UtcNow; + var currentOwner = _store.Owner is not null && now < _store.ExpiresUtc ? _store.Owner : null; + + if (currentOwner is null) + { + _store.Owner = _podId; + _store.ExpiresUtc = now + _ttl; + return Task.FromResult(true); + } + + if (string.Equals(currentOwner, _podId, StringComparison.Ordinal)) + { + _store.ExpiresUtc = now + _ttl; + return Task.FromResult(true); + } + + return Task.FromResult(false); + } + } +} diff --git a/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScheduledTaskWorkerLeaderGatingTests.cs b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScheduledTaskWorkerLeaderGatingTests.cs new file mode 100644 index 000000000..38ab040e4 --- /dev/null +++ b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScheduledTaskWorkerLeaderGatingTests.cs @@ -0,0 +1,189 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using Emby.Server.Implementations.ScheduledTasks; +using MediaBrowser.Common.Configuration; +using MediaBrowser.Controller.ScheduledTasks; +using MediaBrowser.Model.Tasks; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace Jellyfin.Server.Implementations.Tests.ScheduledTasks; + +/// +/// Tests that gates the periodic (timer-driven) execution of gated +/// tasks to the scan leader, while leaving non-gated tasks and manual/API-triggered runs unaffected. +/// +public class ScheduledTaskWorkerLeaderGatingTests +{ + private const string GatedKey = "RefreshLibrary"; + private const string NonGatedKey = "DeleteTranscodeFiles"; + + /// + /// A non-leader must not enqueue a gated task when its periodic trigger fires. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task PeriodicTrigger_NonLeader_GatedTask_DoesNotQueue() + { + var taskManager = new Mock(); + var lease = CreateLease(isLeader: false); + var options = CreateOptions(GatedKey); + var task = new StubScheduledTask(GatedKey); + + using var worker = new ScheduledTaskWorker(task, CreateAppPaths(), taskManager.Object, NullLogger.Instance, lease.Object, options); + + await FireTriggerAsync(worker); + + taskManager.Verify(t => t.QueueScheduledTask(It.IsAny(), It.IsAny()), Times.Never); + lease.Verify(l => l.TryAcquireOrRenewAsync(It.IsAny()), Times.Once); + } + + /// + /// The leader must enqueue a gated task when its periodic trigger fires. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task PeriodicTrigger_Leader_GatedTask_Queues() + { + var taskManager = new Mock(); + var lease = CreateLease(isLeader: true); + var options = CreateOptions(GatedKey); + var task = new StubScheduledTask(GatedKey); + + using var worker = new ScheduledTaskWorker(task, CreateAppPaths(), taskManager.Object, NullLogger.Instance, lease.Object, options); + + await FireTriggerAsync(worker); + + taskManager.Verify(t => t.QueueScheduledTask(task, It.IsAny()), Times.Once); + } + + /// + /// A non-gated task must always enqueue when its periodic trigger fires, even for a non-leader, and + /// must not consult the scan-leader lease at all. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task PeriodicTrigger_NonLeader_NonGatedTask_Queues() + { + var taskManager = new Mock(); + var lease = CreateLease(isLeader: false); + var options = CreateOptions(GatedKey); + var task = new StubScheduledTask(NonGatedKey); + + using var worker = new ScheduledTaskWorker(task, CreateAppPaths(), taskManager.Object, NullLogger.Instance, lease.Object, options); + + await FireTriggerAsync(worker); + + taskManager.Verify(t => t.QueueScheduledTask(task, It.IsAny()), Times.Once); + lease.Verify(l => l.TryAcquireOrRenewAsync(It.IsAny()), Times.Never); + } + + /// + /// A manual/API-triggered run goes through , which must run + /// the task regardless of leadership and must not consult the scan-leader lease. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task Execute_NonLeader_GatedTask_RunsAndIgnoresLease() + { + var realTaskManager = new TaskManager(CreateAppPaths(), new Mock>().Object); + var lease = CreateLease(isLeader: false); + var options = CreateOptions(GatedKey); + var task = new StubScheduledTask(GatedKey); + + using var worker = new ScheduledTaskWorker(task, CreateAppPaths(), realTaskManager, NullLogger.Instance, lease.Object, options); + + await worker.Execute(new TaskOptions()); + + Assert.Equal(1, task.ExecuteCount); + lease.Verify(l => l.TryAcquireOrRenewAsync(It.IsAny()), Times.Never); + } + + private static Mock CreateLease(bool isLeader) + { + var lease = new Mock(); + lease + .Setup(l => l.TryAcquireOrRenewAsync(It.IsAny())) + .ReturnsAsync(isLeader); + return lease; + } + + private static ScanLeaderOptions CreateOptions(params string[] gatedKeys) + => new ScanLeaderOptions { Enabled = true, LeaseDurationSeconds = 60, GatedTaskKeys = gatedKeys }; + + private static IApplicationPaths CreateAppPaths() + { + var dir = Path.Combine(Path.GetTempPath(), "jf-scanleader-tests", Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(dir); + + var appPaths = new Mock(); + appPaths.Setup(p => p.DataPath).Returns(dir); + appPaths.Setup(p => p.ConfigurationDirectoryPath).Returns(dir); + return appPaths.Object; + } + + private static async Task FireTriggerAsync(ScheduledTaskWorker worker) + { + var method = typeof(ScheduledTaskWorker).GetMethod( + "OnTriggerTriggered", + BindingFlags.NonPublic | BindingFlags.Instance); + Assert.NotNull(method); + + method!.Invoke(worker, new object[] { new RecordingTrigger(), EventArgs.Empty }); + + // OnTriggerTriggered is async void; the queue decision completes synchronously against the mocked + // lease, so a short delay lets any continuation settle before the assertion. + await Task.Delay(100); + } + + private sealed class StubScheduledTask : IScheduledTask + { + private readonly string _key; + + public StubScheduledTask(string key) + { + _key = key; + } + + public int ExecuteCount { get; private set; } + + public string Name => "Stub Task"; + + public string Key => _key; + + public string Description => "Stub task for gating tests."; + + public string Category => "Tests"; + + public Task ExecuteAsync(IProgress progress, CancellationToken cancellationToken) + { + ExecuteCount++; + return Task.CompletedTask; + } + + public IEnumerable GetDefaultTriggers() => Array.Empty(); + } + + private sealed class RecordingTrigger : ITaskTrigger + { +#pragma warning disable CS0067 // Required by the interface but unused in this test double. + public event EventHandler? Triggered; +#pragma warning restore CS0067 + + public TaskOptions TaskOptions { get; } = new TaskOptions(); + + public void Start(TaskResult? lastResult, ILogger logger, string taskName, bool isApplicationStartup) + { + } + + public void Stop() + { + } + } +} -- 2.47.3