diff --git a/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs b/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs index bd82faf49..694315b8d 100644 --- a/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs +++ b/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs @@ -1,4 +1,6 @@ using System; +using System.Collections.Generic; +using System.Linq; using System.Text.Json; using System.Threading; using System.Threading.Tasks; @@ -35,6 +37,7 @@ session['LeaseExpiresUtc'] = newTicks redis.call('SET', KEYS[1], cjson.encode(session), 'PX', leaseDurationMs) return 1"; + private readonly IConnectionMultiplexer _redis; private readonly IDatabase _db; private readonly TranscodeStoreOptions _options; private readonly ILogger _logger; @@ -50,6 +53,7 @@ return 1"; IOptions options, ILogger logger) { + _redis = redis; _db = redis.GetDatabase(); _options = options.Value; _logger = logger; @@ -140,4 +144,54 @@ return 1"; } private static string GetKey(string playSessionId) => KeyPrefix + playSessionId; + + /// + public async Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + var sessions = new List(); + var servers = _redis.GetServers(); + + foreach (var server in servers) + { + if (!server.IsConnected) + { + continue; + } + + var keys = new List(); + await foreach (var key in server.KeysAsync(database: _db.Database, pattern: KeyPrefix + "*", pageSize: 1000).WithCancellation(cancellationToken).ConfigureAwait(false)) + { + keys.Add(key); + } + + var tasks = keys.Select(key => _db.StringGetAsync(key)).ToList(); + var values = await Task.WhenAll(tasks).ConfigureAwait(false); + + foreach (var raw in values) + { + if (!raw.HasValue) + { + continue; + } + + TranscodeSession? session; + try + { + session = JsonSerializer.Deserialize(raw.ToString()); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to deserialize transcode session from Redis."); + continue; + } + + if (session is not null) + { + sessions.Add(session); + } + } + } + + return sessions; + } } diff --git a/Emby.Server.Implementations/ScheduledTasks/Tasks/DeleteTranscodeFileTask.cs b/Emby.Server.Implementations/ScheduledTasks/Tasks/DeleteTranscodeFileTask.cs index 9cc2cc512..5a7142986 100644 --- a/Emby.Server.Implementations/ScheduledTasks/Tasks/DeleteTranscodeFileTask.cs +++ b/Emby.Server.Implementations/ScheduledTasks/Tasks/DeleteTranscodeFileTask.cs @@ -5,6 +5,7 @@ using System.Threading; using System.Threading.Tasks; using MediaBrowser.Common.Configuration; using MediaBrowser.Controller.IO; +using MediaBrowser.Controller.MediaEncoding; using MediaBrowser.Model.Globalization; using MediaBrowser.Model.IO; using MediaBrowser.Model.Tasks; @@ -21,6 +22,7 @@ public class DeleteTranscodeFileTask : IScheduledTask, IConfigurableScheduledTas private readonly IConfigurationManager _configurationManager; private readonly IFileSystem _fileSystem; private readonly ILocalizationManager _localization; + private readonly ITranscodeSessionStore _sessionStore; /// /// Initializes a new instance of the class. @@ -29,16 +31,19 @@ public class DeleteTranscodeFileTask : IScheduledTask, IConfigurableScheduledTas /// Instance of the interface. /// Instance of the interface. /// Instance of the interface. + /// Instance of the interface. public DeleteTranscodeFileTask( ILogger logger, IFileSystem fileSystem, IConfigurationManager configurationManager, - ILocalizationManager localization) + ILocalizationManager localization, + ITranscodeSessionStore sessionStore) { _logger = logger; _fileSystem = fileSystem; _configurationManager = configurationManager; _localization = localization; + _sessionStore = sessionStore; } /// @@ -78,25 +83,39 @@ public class DeleteTranscodeFileTask : IScheduledTask, IConfigurableScheduledTas } /// - public Task ExecuteAsync(IProgress progress, CancellationToken cancellationToken) + public async Task ExecuteAsync(IProgress progress, CancellationToken cancellationToken) { var minDateModified = DateTime.UtcNow.AddDays(-1); progress.Report(50); - DeleteTempFilesFromDirectory(_configurationManager.GetTranscodePath(), minDateModified, progress, cancellationToken); + IEnumerable activeSessions; + try + { + activeSessions = await _sessionStore.GetActiveSessionsAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to retrieve active transcode sessions. Skipping deletion to avoid removing files in use."); + progress.Report(100); + return; + } - return Task.CompletedTask; + DeleteTempFilesFromDirectory(_configurationManager.GetTranscodePath(), minDateModified, activeSessions, progress, cancellationToken); } /// - /// Deletes the transcoded temp files from directory with a last write time less than a given date. + /// Deletes the transcoded temp files from directory with a last write time less than a given date, + /// skipping any files that belong to an active transcode session. /// /// The directory. /// The min date modified. + /// The currently active transcode sessions. /// The progress. /// The task cancellation token. - private void DeleteTempFilesFromDirectory(string directory, DateTime minDateModified, IProgress progress, CancellationToken cancellationToken) + private void DeleteTempFilesFromDirectory(string directory, DateTime minDateModified, IEnumerable activeSessions, IProgress progress, CancellationToken cancellationToken) { + var activeSessionList = activeSessions.ToList(); + var filesToDelete = _fileSystem.GetFiles(directory, true) .Where(f => _fileSystem.GetLastWriteTimeUtc(f) < minDateModified) .ToList(); @@ -112,6 +131,13 @@ public class DeleteTranscodeFileTask : IScheduledTask, IConfigurableScheduledTas cancellationToken.ThrowIfCancellationRequested(); + if (IsFileProtectedByActiveSession(file.FullName, activeSessionList)) + { + _logger.LogDebug("Skipping deletion of {FilePath} as it belongs to an active transcode session.", file.FullName); + index++; + continue; + } + FileSystemHelper.DeleteFile(_fileSystem, file.FullName, _logger); index++; @@ -121,4 +147,24 @@ public class DeleteTranscodeFileTask : IScheduledTask, IConfigurableScheduledTas progress.Report(100); } + + private static bool IsFileProtectedByActiveSession(string filePath, IList activeSessions) + { + foreach (var session in activeSessions) + { + if (!string.IsNullOrEmpty(session.ManifestPath) && + string.Equals(filePath, session.ManifestPath, StringComparison.OrdinalIgnoreCase)) + { + return true; + } + + if (!string.IsNullOrEmpty(session.SegmentPathPrefix) && + filePath.StartsWith(session.SegmentPathPrefix, StringComparison.OrdinalIgnoreCase)) + { + return true; + } + } + + return false; + } } diff --git a/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs b/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs index 9ab00f70a..9ae802731 100644 --- a/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs +++ b/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs @@ -1,3 +1,4 @@ +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -59,4 +60,14 @@ public interface ITranscodeSessionStore /// A cancellation token. /// A representing the asynchronous operation. Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default); + + /// + /// Returns all currently active transcoding sessions from the store. + /// + /// A cancellation token. + /// + /// An enumerable of objects representing all active sessions. + /// Returns an empty enumerable if no sessions are active or if the store cannot be reached. + /// + Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default); } diff --git a/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs b/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs index e7dcdd1d1..4626a677d 100644 --- a/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs +++ b/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs @@ -1,3 +1,5 @@ +using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -28,4 +30,8 @@ public sealed class NullTranscodeSessionStore : ITranscodeSessionStore /// public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) => Task.CompletedTask; + + /// + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + => Task.FromResult>(Array.Empty()); } diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs index 23b231e11..e0b0f492c 100644 --- a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs +++ b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Jellyfin.Api.Controllers; @@ -172,6 +173,18 @@ namespace Jellyfin.Api.Tests.Controllers return Task.CompletedTask; } + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + lock (_lock) + { + var sessions = _sessions.Values + .Where(s => s.LeaseExpiresUtc > DateTime.UtcNow) + .Select(Clone) + .ToList(); + return Task.FromResult>(sessions); + } + } + private static TranscodeSession Clone(TranscodeSession source) => new TranscodeSession { diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs index 05a5e3948..1c7d9dbd2 100644 --- a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs +++ b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Controller.MediaEncoding; @@ -184,6 +185,18 @@ namespace Jellyfin.Api.Tests.Controllers return Task.CompletedTask; } + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + lock (_lock) + { + var sessions = _sessions.Values + .Where(s => s.LeaseExpiresUtc > DateTime.UtcNow) + .Select(Clone) + .ToList(); + return Task.FromResult>(sessions); + } + } + private static TranscodeSession Clone(TranscodeSession source) => new TranscodeSession { diff --git a/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs b/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs index 82772ca67..0d9c4d00d 100644 --- a/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs +++ b/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Controller.MediaEncoding; @@ -92,6 +93,16 @@ public sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore return Task.CompletedTask; } + /// + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + lock (_lock) + { + var sessions = _sessions.Values.Select(Clone).ToList(); + return Task.FromResult>(sessions); + } + } + private static TranscodeSession Clone(TranscodeSession source) => new TranscodeSession { diff --git a/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs b/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs index 3bc51b002..873fb5662 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Controller.MediaEncoding; @@ -209,6 +210,19 @@ public class RedisTranscodeSessionStoreTests return Task.CompletedTask; } + /// + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + lock (_lock) + { + var sessions = _sessions.Values + .Where(s => s.LeaseExpiresUtc > DateTime.UtcNow) + .Select(Clone) + .ToList(); + return Task.FromResult>(sessions); + } + } + private static TranscodeSession Clone(TranscodeSession source) => new TranscodeSession { diff --git a/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs index 30d6b7470..90dd010d5 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs @@ -1,8 +1,13 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; +using MediaBrowser.Common.Configuration; using MediaBrowser.Controller.MediaEncoding; +using MediaBrowser.Model.Configuration; +using MediaBrowser.Model.IO; +using Moq; using Xunit; namespace Jellyfin.Server.Implementations.Tests.ScheduledTasks; @@ -31,6 +36,27 @@ public class DeleteTranscodeFileTaskTests LastDurablePlaybackOffset = 12_000_000L, }; + /// + /// Creates a mock that returns + /// as the configured transcode path, used by the GetTranscodePath extension method. + /// + private static Mock CreateConfigMock(string transcodePath) + { + var appPathsMock = new Mock(); + appPathsMock + .Setup(p => p.CreateAndCheckMarker(It.IsAny(), It.IsAny(), It.IsAny())); + + var configMock = new Mock(); + configMock + .Setup(c => c.GetConfiguration("encoding")) + .Returns(new EncodingOptions { TranscodingTempPath = transcodePath }); + configMock + .Setup(c => c.CommonApplicationPaths) + .Returns(appPathsMock.Object); + + return configMock; + } + /// /// A directory that belongs to a session with a live lease must NOT be deleted. /// The store returns non-null, signalling to the cleanup task that the session is active. @@ -87,6 +113,206 @@ public class DeleteTranscodeFileTaskTests Assert.Null(liveSession); } + /// + /// Files that belong to an active session (manifest or segments) must NOT be deleted + /// even when their modification time is older than minDateModified. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task ExecuteAsync_WithActiveSession_DoesNotDeleteActiveFiles() + { + // Arrange + const string TranscodePath = "/transcode"; + const string SessionId = "active-session-1"; + const string ManifestPath = "/transcode/active-session-1/manifest.m3u8"; + const string SegmentPath = "/transcode/active-session-1/segment0.ts"; + + var store = new CleanupTestSessionStore(); + var session = new TranscodeSession + { + PlaySessionId = SessionId, + OwnerPod = "pod-a", + LeaseExpiresUtc = DateTime.UtcNow.AddMinutes(5), + ManifestPath = ManifestPath, + SegmentPathPrefix = "/transcode/active-session-1/segment", + MediaSourceId = "media-source-1", + }; + await store.SetAsync(session); + + var deletedFiles = new List(); + var oldModifyTime = DateTime.UtcNow.AddDays(-2); + + var fileSystemMock = new Mock(); + fileSystemMock + .Setup(fs => fs.GetFiles(TranscodePath, true)) + .Returns(new[] + { + new FileSystemMetadata { FullName = ManifestPath, IsDirectory = false }, + new FileSystemMetadata { FullName = SegmentPath, IsDirectory = false }, + }); + fileSystemMock + .Setup(fs => fs.GetLastWriteTimeUtc(It.IsAny())) + .Returns(oldModifyTime); + fileSystemMock + .Setup(fs => fs.DeleteFile(It.IsAny())) + .Callback(path => deletedFiles.Add(path)); + fileSystemMock + .Setup(fs => fs.GetFiles(TranscodePath, false)) + .Returns(Enumerable.Empty()); + fileSystemMock + .Setup(fs => fs.GetDirectories(It.IsAny(), It.IsAny())) + .Returns(Enumerable.Empty()); + + var configMock = CreateConfigMock(TranscodePath); + + var localizationMock = new Mock(); + localizationMock + .Setup(l => l.GetLocalizedString(It.IsAny())) + .Returns(s => s); + + var loggerMock = new Mock>(); + + var task = new Emby.Server.Implementations.ScheduledTasks.Tasks.DeleteTranscodeFileTask( + loggerMock.Object, + fileSystemMock.Object, + configMock.Object, + localizationMock.Object, + store); + + // Act + await task.ExecuteAsync(new Progress(), CancellationToken.None); + + // Assert – neither the manifest nor the segment should have been deleted + Assert.DoesNotContain(ManifestPath, deletedFiles); + Assert.DoesNotContain(SegmentPath, deletedFiles); + } + + /// + /// Files whose session lease has expired are NOT returned by + /// and therefore should be eligible for time-based deletion. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task ExecuteAsync_WithExpiredSession_DeletesFiles() + { + // Arrange + const string TranscodePath = "/transcode"; + const string SessionId = "expired-session-1"; + const string ManifestPath = "/transcode/expired-session-1/manifest.m3u8"; + + var store = new CleanupTestSessionStore(); + // Lease expired two hours ago + var session = new TranscodeSession + { + PlaySessionId = SessionId, + OwnerPod = "pod-a", + LeaseExpiresUtc = DateTime.UtcNow.AddHours(-2), + ManifestPath = ManifestPath, + SegmentPathPrefix = "/transcode/expired-session-1/segment", + MediaSourceId = "media-source-1", + }; + await store.SetAsync(session); + + var deletedFiles = new List(); + var oldModifyTime = DateTime.UtcNow.AddDays(-2); + + var fileSystemMock = new Mock(); + fileSystemMock + .Setup(fs => fs.GetFiles(TranscodePath, true)) + .Returns(new[] + { + new FileSystemMetadata { FullName = ManifestPath, IsDirectory = false }, + }); + fileSystemMock + .Setup(fs => fs.GetLastWriteTimeUtc(It.IsAny())) + .Returns(oldModifyTime); + fileSystemMock + .Setup(fs => fs.DeleteFile(It.IsAny())) + .Callback(path => deletedFiles.Add(path)); + fileSystemMock + .Setup(fs => fs.GetDirectories(It.IsAny(), It.IsAny())) + .Returns(Enumerable.Empty()); + + var configMock = CreateConfigMock(TranscodePath); + + var localizationMock = new Mock(); + localizationMock + .Setup(l => l.GetLocalizedString(It.IsAny())) + .Returns(s => s); + + var loggerMock = new Mock>(); + + var task = new Emby.Server.Implementations.ScheduledTasks.Tasks.DeleteTranscodeFileTask( + loggerMock.Object, + fileSystemMock.Object, + configMock.Object, + localizationMock.Object, + store); + + // Act + await task.ExecuteAsync(new Progress(), CancellationToken.None); + + // Assert – expired session files are eligible for time-based deletion + Assert.Contains(ManifestPath, deletedFiles); + } + + /// + /// When throws an exception, + /// the task should abort deletion safely rather than risk removing files in use. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task ExecuteAsync_WhenStoreFails_AbortsDeletion() + { + // Arrange + const string TranscodePath = "/transcode"; + const string ManifestPath = "/transcode/session-1/manifest.m3u8"; + + var deletedFiles = new List(); + var oldModifyTime = DateTime.UtcNow.AddDays(-2); + + var fileSystemMock = new Mock(); + fileSystemMock + .Setup(fs => fs.GetFiles(TranscodePath, true)) + .Returns(new[] + { + new FileSystemMetadata { FullName = ManifestPath, IsDirectory = false }, + }); + fileSystemMock + .Setup(fs => fs.GetLastWriteTimeUtc(It.IsAny())) + .Returns(oldModifyTime); + fileSystemMock + .Setup(fs => fs.DeleteFile(It.IsAny())) + .Callback(path => deletedFiles.Add(path)); + + var configMock = CreateConfigMock(TranscodePath); + + var localizationMock = new Mock(); + localizationMock + .Setup(l => l.GetLocalizedString(It.IsAny())) + .Returns(s => s); + + var loggerMock = new Mock>(); + + var failingStoreMock = new Mock(); + failingStoreMock + .Setup(s => s.GetActiveSessionsAsync(It.IsAny())) + .ThrowsAsync(new InvalidOperationException("Redis unavailable")); + + var task = new Emby.Server.Implementations.ScheduledTasks.Tasks.DeleteTranscodeFileTask( + loggerMock.Object, + fileSystemMock.Object, + configMock.Object, + localizationMock.Object, + failingStoreMock.Object); + + // Act + await task.ExecuteAsync(new Progress(), CancellationToken.None); + + // Assert – when the store fails, no files should be deleted (safe abort) + Assert.Empty(deletedFiles); + } + /// /// Minimal in-memory used within this test class /// to avoid a cross-project reference to Jellyfin.MediaEncoding.Tests. @@ -166,6 +392,18 @@ public class DeleteTranscodeFileTaskTests return Task.CompletedTask; } + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + lock (_lock) + { + var sessions = _sessions.Values + .Where(s => s.LeaseExpiresUtc > DateTime.UtcNow) + .Select(Clone) + .ToList(); + return Task.FromResult>(sessions); + } + } + private static TranscodeSession Clone(TranscodeSession source) => new TranscodeSession {