Fix concurrent ffmpeg segment racing

This is a nasty one. The failure mode is:

1. Request A started FFmpeg and waited for a segment.
2. Request B requested an earlier or far away segment.
3. Jellyfin thought FFmpeg should to restart at a different position.
4. Request B killed the existing transcoding job.
5. Killing that job cancelled the same token request A was using.
6. The cancellation produced http 500 to request A.

To fix this:

we lock transcoding job state changes and segment handling per playlist, and use a thread safe counter to track how many http responses are still using each job’s segments. A job is only stopped or replaced once that counter reaches zero.

1 files changed, 76 insertions, 0 deletions

diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsControllerTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsControllerTests.cs
index 1f06e8fde6..5f5f273f12 100644
--- a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsControllerTests.cs
+++ b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsControllerTests.cs
@@ -1,5 +1,9 @@
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 using Jellyfin.Api.Controllers;
+using MediaBrowser.Controller.MediaEncoding;
+using Microsoft.Extensions.Logging.Abstractions;
 using Xunit;
 
 namespace Jellyfin.Api.Tests.Controllers
@@ -41,5 +45,77 @@ namespace Jellyfin.Api.Tests.Controllers
 
             return data;
         }
+
+        [Fact]
+        public async Task WaitForActiveTranscodingRequests_WaitsUntilRequestCompletes()
+        {
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance)
+            {
+                ActiveRequestCount = 1
+            };
+
+            var waitTask = DynamicHlsController.WaitForActiveTranscodingRequests(job, CancellationToken.None);
+            Assert.False(waitTask.IsCompleted);
+
+            job.DecrementActiveRequestCount();
+
+            await waitTask;
+        }
+
+        [Fact]
+        public async Task WaitForActiveTranscodingRequests_WaitsForEveryRequest()
+        {
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance)
+            {
+                ActiveRequestCount = 2
+            };
+
+            var waitTask = DynamicHlsController.WaitForActiveTranscodingRequests(job, CancellationToken.None);
+            job.DecrementActiveRequestCount();
+
+            await Task.Delay(150, TestContext.Current.CancellationToken);
+            Assert.False(waitTask.IsCompleted);
+
+            job.DecrementActiveRequestCount();
+
+            await waitTask;
+        }
+
+        [Fact]
+        public async Task WaitForActiveTranscodingRequests_ReturnsWithoutAnActiveRequest()
+        {
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance);
+
+            await DynamicHlsController.WaitForActiveTranscodingRequests(job, CancellationToken.None);
+            await DynamicHlsController.WaitForActiveTranscodingRequests(null, CancellationToken.None);
+        }
+
+        [Fact]
+        public async Task WaitForActiveTranscodingRequests_ObservesCancellation()
+        {
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance)
+            {
+                ActiveRequestCount = 1
+            };
+            using var cancellationTokenSource = new CancellationTokenSource();
+
+            var waitTask = DynamicHlsController.WaitForActiveTranscodingRequests(job, cancellationTokenSource.Token);
+            await cancellationTokenSource.CancelAsync();
+
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(() => waitTask);
+        }
+
+        [Fact]
+        public async Task ActiveRequestCount_UpdatesAtomically()
+        {
+            const int RequestCount = 1000;
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance);
+
+            await Task.WhenAll(
+                Task.Run(() => Parallel.For(0, RequestCount, _ => job.IncrementActiveRequestCount())),
+                Task.Run(() => Parallel.For(0, RequestCount, _ => job.DecrementActiveRequestCount())));
+
+            Assert.Equal(0, job.ActiveRequestCount);
+        }
     }
 }