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() + { + } + } +}