From c43992630ea8265f18f4d4a803ca4016b9547920 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Fri, 11 Sep 2026 23:56:35 +1000 Subject: [PATCH] feat: skip transcode files held by an active session Every replica runs the transcode cleanup task against the same shared transcode directory, so one pod deletes segments another pod is still streaming. - read the active sessions from ITranscodeSessionStore before deleting - keep files matching an active session manifest path or segment prefix - abort the sweep when the store cannot be reached --- .../Tasks/DeleteTranscodeFileTask.cs | 58 ++- .../DeleteTranscodeFileTaskTests.cs | 429 ++++++++++++++++++ 2 files changed, 481 insertions(+), 6 deletions(-) create mode 100644 tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs diff --git a/Emby.Server.Implementations/ScheduledTasks/Tasks/DeleteTranscodeFileTask.cs b/Emby.Server.Implementations/ScheduledTasks/Tasks/DeleteTranscodeFileTask.cs index 9cc2cc5123..5a71429864 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/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs new file mode 100644 index 0000000000..0abf2f5e3d --- /dev/null +++ b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs @@ -0,0 +1,429 @@ +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; + +/// +/// Tests for lease-aware cleanup behavior expected of DeleteTranscodeFileTask once +/// it is made HA-aware in Phase 5.2. +/// +/// The current DeleteTranscodeFileTask implementation uses file-age only and does not +/// check , which creates a data-loss risk on shared NFS +/// storage. These tests document the correct contract by exercising the store directly. +/// +/// +public class DeleteTranscodeFileTaskTests +{ + private static TranscodeSession CreateSession(string id, string pod, DateTime leaseExpiry) + => new TranscodeSession + { + PlaySessionId = id, + OwnerPod = pod, + LeaseExpiresUtc = leaseExpiry, + ManifestPath = $"/transcode/{id}/manifest.m3u8", + SegmentPathPrefix = $"/transcode/{id}/segment", + MediaSourceId = $"media-source-{id}", + LastCompletedSegmentIndex = 2, + 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. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task LiveLease_StoreReturnsSession_DirectoryShouldNotBeDeleted() + { + var store = new CleanupTestSessionStore(); + var session = CreateSession("cleanup-session-1", "pod-a", DateTime.UtcNow.AddMinutes(5)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + // The cleanup task should query the store before deleting. + var liveSession = await store.TryGetAsync("cleanup-session-1", TestContext.Current.CancellationToken); + + // Non-null result → lease is active → directory must be retained. + Assert.NotNull(liveSession); + Assert.Equal("pod-a", liveSession.OwnerPod); + Assert.True(liveSession.LeaseExpiresUtc > DateTime.UtcNow); + } + + /// + /// A directory whose session lease has expired beyond the recovery window MAY be deleted. + /// The store returns null, signalling to the cleanup task that deletion is safe. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task ExpiredBeyondRecoveryWindow_StoreReturnsNull_DirectoryMayBeDeleted() + { + var store = new CleanupTestSessionStore(); + + // Lease expired two hours ago – beyond any reasonable recovery window. + var session = CreateSession("cleanup-session-2", "pod-a", DateTime.UtcNow.AddHours(-2)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + var liveSession = await store.TryGetAsync("cleanup-session-2", TestContext.Current.CancellationToken); + + // Null result → lease is expired → cleanup task may delete the directory. + Assert.Null(liveSession); + } + + /// + /// When no session record exists in the store for a given directory, the cleanup task + /// should treat the directory as deletable (store returns null). + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task NoSessionRecord_StoreReturnsNull_DirectoryMayBeDeleted() + { + var store = new CleanupTestSessionStore(); + + var liveSession = await store.TryGetAsync("unknown-session", TestContext.Current.CancellationToken); + + 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, TestContext.Current.CancellationToken); + + 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, TestContext.Current.CancellationToken); + + 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. + /// + private sealed class CleanupTestSessionStore : ITranscodeSessionStore + { + private static readonly TimeSpan LeaseDuration = TimeSpan.FromSeconds(30); + + private readonly Dictionary _sessions = + new(StringComparer.OrdinalIgnoreCase); + + private readonly Lock _lock = new(); + + public Task TryGetAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (_sessions.TryGetValue(playSessionId, out var s) && s.LeaseExpiresUtc > DateTime.UtcNow) + { + return Task.FromResult(Clone(s)); + } + + return Task.FromResult(null); + } + } + + public Task TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (!_sessions.TryGetValue(playSessionId, out var s)) + { + return Task.FromResult(false); + } + + if (s.LeaseExpiresUtc > DateTime.UtcNow) + { + return Task.FromResult(false); + } + + s.OwnerPod = claimingPod; + s.LeaseExpiresUtc = DateTime.UtcNow.Add(LeaseDuration); + return Task.FromResult(true); + } + } + + public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _sessions[session.PlaySessionId] = session; + } + + return Task.CompletedTask; + } + + public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (_sessions.TryGetValue(playSessionId, out var s)) + { + s.LeaseExpiresUtc = DateTime.UtcNow.Add(LeaseDuration); + } + } + + return Task.CompletedTask; + } + + public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _sessions.Remove(playSessionId); + } + + 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); + } + } + + public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + public Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + => Task.FromResult(null); + + public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + private static TranscodeSession Clone(TranscodeSession source) + => new TranscodeSession + { + PlaySessionId = source.PlaySessionId, + OwnerPod = source.OwnerPod, + LeaseExpiresUtc = source.LeaseExpiresUtc, + ManifestPath = source.ManifestPath, + SegmentPathPrefix = source.SegmentPathPrefix, + MediaSourceId = source.MediaSourceId, + LastCompletedSegmentIndex = source.LastCompletedSegmentIndex, + LastDurablePlaybackOffset = source.LastDurablePlaybackOffset, + }; + } +}