diff --git a/CHANGELOG.md b/CHANGELOG.md index 27d865e45..a80be2592 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,9 +25,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). - Those with customized `episode.sbntxt` templates will want to make a similar fix - Disable HDHR endpoints when JWT is used; they never worked in this configuration in the first place - Fix scanners deleting, duplicating and stranding each other's tags; tags now track which scanner owns them - - Symptoms included a show losing every tag when it changed networks (dropping it from collections and schedules), networks vanishing after a library scan, and labels or collections that were renamed or deleted on the media server never going away in ETV - - To recover a show that already lost its tags: deep scan it from the show page, then use **Deep Scan Collections** on the libraries page; networks return on the next network scan with no action needed + - Symptoms included: + - A show losing every tag when it changed networks (dropping it from collections and schedules) + - Networks vanishing after a library scan + - Labels or collections that were renamed or deleted on the media server never going away in ETV + - To recover a show that already lost its tags: deep scan it from the show page, then use **Deep Scan Collections** on the libraries page - `` from NFO metadata is now searchable with `country:` instead of `tag:`; deep scan a local library to convert existing items +- Fix multiple channel startup failure causes ## [26.8.1] - 2026-08-29 ### Security diff --git a/ErsatzTV.Application/Streaming/Commands/StartFFmpegNextSessionHandler.cs b/ErsatzTV.Application/Streaming/Commands/StartFFmpegNextSessionHandler.cs index 3f9d855cf..98b3dd23b 100644 --- a/ErsatzTV.Application/Streaming/Commands/StartFFmpegNextSessionHandler.cs +++ b/ErsatzTV.Application/Streaming/Commands/StartFFmpegNextSessionHandler.cs @@ -16,7 +16,6 @@ using ErsatzTV.Core.Next.Config; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; -using Subtitle = ErsatzTV.Core.Next.Config.Subtitle; namespace ErsatzTV.Application.Streaming; @@ -34,31 +33,64 @@ public class StartFFmpegNextSessionHandler( ILogger sessionWorkerLogger) : NextChannelHandlerBase(fileSystem), IRequestHandler> { + private static readonly TimeSpan StartDeadline = TimeSpan.FromSeconds(30); + private readonly IFileSystem _fileSystem = fileSystem; - public Task> Handle( - StartFFmpegNextSession request, - CancellationToken cancellationToken) => - Validate(request, cancellationToken) - .MapT(validationResult => StartProcess(request, validationResult, cancellationToken)) - // this weirdness is needed to maintain the error type (.ToEitherAsync() just gives BaseError) -#pragma warning disable VSTHRD103 - .Bind(v => v.ToEither().MapLeft(seq => seq.Head()).MapAsync, string>(identity)); -#pragma warning restore VSTHRD103 - - private async Task StartProcess( + public async Task> Handle( StartFFmpegNextSession request, - ValidationResult validationResult, CancellationToken cancellationToken) { + using IDisposable releaser = + await ffmpegSegmenterService.LockForStart(request.ChannelNumber, cancellationToken); + + if (ffmpegSegmenterService.TryGetWorker(request.ChannelNumber, out IHlsSessionWorker existing)) + { + existing.Touch(Option.None); + return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request)); + } + + Validation maybeChannelBinary = await ChannelBinaryMustExist(); + if (maybeChannelBinary.IsFail) + { + return maybeChannelBinary.FailToSeq().Head(); + } + + string channelBinary = maybeChannelBinary.SuccessToSeq().Head(); + Option idleTimeout = Option.None; // Option targetFramerate = await mediator.Send( // new GetChannelFramerate(request.ChannelNumber), // cancellationToken); + int initialSegmentCount = await configElementRepository + .GetValue(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken) + .Map(maybeCount => maybeCount.Match(identity, () => 1)); + + Option maybeChannel = + await mediator.Send(new GetChannelByNumber(request.ChannelNumber), cancellationToken); + + if (maybeChannel.IsNone) + { + return BaseError.New($"Channel number {request.ChannelNumber} does not exist."); + } + + ChannelViewModel channel = maybeChannel.Head(); + + Option maybeFFmpegProfile = await mediator.Send( + new GetFFmpegProfileById(channel.FFmpegProfileId), + cancellationToken); + + if (maybeFFmpegProfile.IsNone) + { + return BaseError.New($"FFmpeg profile {channel.FFmpegProfileId} not exist"); + } + + FFmpegProfileViewModel ffmpegProfile = maybeFFmpegProfile.Head(); + // only load timeout when needed - if (validationResult.Channel.IdleBehavior is not ChannelIdleBehavior.KeepRunning) + if (channel.IdleBehavior is not ChannelIdleBehavior.KeepRunning) { idleTimeout = await configElementRepository .GetValue(ConfigElementKey.FFmpegSegmenterTimeout, cancellationToken) @@ -67,111 +99,53 @@ public class StartFFmpegNextSessionHandler( await mediator.Send(new RefreshGraphicsElements(), cancellationToken); - ChannelConfig config = await channelConfigConverter.ToNext( - validationResult.Channel, - validationResult.FfmpegProfile, - cancellationToken); + PrepareTranscodeFolder(request.ChannelNumber); + + ChannelConfig config = await channelConfigConverter.ToNext(channel, ffmpegProfile, cancellationToken); NextSessionWorker worker = new NextSessionWorker( - validationResult.ChannelBinary, + channelBinary, config, _fileSystem, localFileSystem, serviceScopeFactory, sessionWorkerLogger); - ffmpegSegmenterService.AddOrUpdateWorker(request.ChannelNumber, worker); + if (!ffmpegSegmenterService.TryAddWorker(request.ChannelNumber, worker)) + { + return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request)); + } // fire and forget worker - _ = worker.Run(request.ChannelNumber, idleTimeout, hostApplicationLifetime.ApplicationStopping) - .ContinueWith( + Task runTask = worker.Run(request.ChannelNumber, idleTimeout, hostApplicationLifetime.ApplicationStopping); + _ = runTask.ContinueWith( _ => { - ffmpegSegmenterService.RemoveWorker(request.ChannelNumber, out IHlsSessionWorker inactiveWorker); + ffmpegSegmenterService.RemoveWorker(request.ChannelNumber, worker); - inactiveWorker?.Dispose(); + ((IDisposable)worker).Dispose(); workerChannel.TryWrite(new ReleaseMemory(false)); }, TaskScheduler.Default); - int initialSegmentCount = await configElementRepository - .GetValue(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken) - .Map(maybeCount => maybeCount.Match(identity, () => 1)); - - await worker.WaitForPlaylistSegments(initialSegmentCount, cancellationToken); - - return await GetMultiVariantPlaylist(request); - } - - private Task> Validate( - StartFFmpegNextSession request, - CancellationToken cancellationToken) => - SessionMustBeInactive(request) - .BindT(_ => FolderMustBeEmpty(request)) - .BindT(_ => ChannelBinaryMustExist()) - .BindT(channelBinary => ChannelMustExist(request, new ValidationResult(channelBinary, null, null), cancellationToken)) - .BindT(result => FFmpegProfileMustExist(result, cancellationToken)); - - private async Task> SessionMustBeInactive(StartFFmpegNextSession request) - { - var result = Optional(ffmpegSegmenterService.TryAddWorker(request.ChannelNumber, null)) - .Where(success => success) - .Map(_ => Unit.Default) - .ToValidation(new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request))); - - if (result.IsFail && ffmpegSegmenterService.TryGetWorker( - request.ChannelNumber, - out IHlsSessionWorker worker)) - { - worker?.Touch(Option.None); - } - - return result; + Either ready = await SessionStartWait.ForReady( + request.ChannelNumber, + worker, + runTask, + initialSegmentCount, + StartDeadline, + cancellationToken); + return await ready.MapAsync(async _ => await GetMultiVariantPlaylist(request)); } - private Task> FolderMustBeEmpty(StartFFmpegNextSession request) + private void PrepareTranscodeFolder(string channelNumber) { - string folder = Path.Combine(FileSystemLayout.TranscodeFolder, request.ChannelNumber); + string folder = Path.Combine(FileSystemLayout.TranscodeFolder, channelNumber); logger.LogDebug("Preparing transcode folder {Folder}", folder); localFileSystem.EnsureFolderExists(folder); localFileSystem.EmptyFolder(folder); - - return Task.FromResult>(Unit.Default); - } - - private async Task> ChannelMustExist( - StartFFmpegNextSession request, - ValidationResult result, - CancellationToken cancellationToken) - { - Option maybeChannel = await mediator.Send( - new GetChannelByNumber(request.ChannelNumber), - cancellationToken); - - foreach (ChannelViewModel channel in maybeChannel) - { - return result with { Channel = channel }; - } - - return BaseError.New($"Channel number {request.ChannelNumber} does not exist"); - } - - private async Task> FFmpegProfileMustExist( - ValidationResult result, - CancellationToken cancellationToken) - { - Option maybeFFmpegProfile = await mediator.Send( - new GetFFmpegProfileById(result.Channel.FFmpegProfileId), - cancellationToken); - - foreach (FFmpegProfileViewModel ffmpegProfile in maybeFFmpegProfile) - { - return result with { FfmpegProfile = ffmpegProfile }; - } - - return BaseError.New($"FFmpeg profile {result.Channel.FFmpegProfileId} not exist"); } private async Task GetMultiVariantPlaylist(StartFFmpegNextSession request) @@ -225,9 +199,4 @@ public class StartFFmpegNextSessionHandler( #EXT-X-STREAM-INF:BANDWIDTH={bitrate}{resolution} {variantPlaylist}"; } - - private sealed record ValidationResult( - string ChannelBinary, - ChannelViewModel Channel, - FFmpegProfileViewModel FfmpegProfile); } diff --git a/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs b/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs index c8322bf96..d646897cb 100644 --- a/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs +++ b/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs @@ -22,6 +22,8 @@ namespace ErsatzTV.Application.Streaming; public class StartFFmpegSessionHandler : IRequestHandler> { + private static readonly TimeSpan StartDeadline = TimeSpan.FromSeconds(30); + private readonly IFileSystem _fileSystem; private readonly IConfigElementRepository _configElementRepository; private readonly IFFmpegSegmenterService _ffmpegSegmenterService; @@ -66,16 +68,17 @@ public class StartFFmpegSessionHandler : IRequestHandler> Handle(StartFFmpegSession request, CancellationToken cancellationToken) => - Validate(request) - .MapT(_ => StartProcess(request, cancellationToken)) - // this weirdness is needed to maintain the error type (.ToEitherAsync() just gives BaseError) -#pragma warning disable VSTHRD103 - .Bind(v => v.ToEither().MapLeft(seq => seq.Head()).MapAsync, string>(identity)); -#pragma warning restore VSTHRD103 - - private async Task StartProcess(StartFFmpegSession request, CancellationToken cancellationToken) + public async Task> Handle(StartFFmpegSession request, CancellationToken cancellationToken) { + using IDisposable releaser = + await _ffmpegSegmenterService.LockForStart(request.ChannelNumber, cancellationToken); + + if (_ffmpegSegmenterService.TryGetWorker(request.ChannelNumber, out IHlsSessionWorker existing)) + { + existing.Touch(Option.None); + return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request)); + } + Option idleTimeout = await _configElementRepository .GetValue(ConfigElementKey.FFmpegSegmenterTimeout, cancellationToken) .Map(maybeTimeout => maybeTimeout.Match(i => TimeSpan.FromSeconds(i), () => TimeSpan.FromMinutes(1))); @@ -84,6 +87,10 @@ public class StartFFmpegSessionHandler : IRequestHandler(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken) + .Map(maybeCount => maybeCount.Match(identity, () => 1)); + // disable idle timeout when configured to keep running Option channel = await _mediator.Send(new GetChannelByNumber(request.ChannelNumber), cancellationToken); @@ -94,30 +101,35 @@ public class StartFFmpegSessionHandler : IRequestHandler - { - _ffmpegSegmenterService.RemoveWorker(request.ChannelNumber, out IHlsSessionWorker inactiveWorker); - - inactiveWorker?.Dispose(); + Task runTask = worker.Run(request.ChannelNumber, idleTimeout, _hostApplicationLifetime.ApplicationStopping); + _ = runTask.ContinueWith( + _ => + { + _ffmpegSegmenterService.RemoveWorker(request.ChannelNumber, worker); - _workerChannel.TryWrite(new ReleaseMemory(false)); - }, - TaskScheduler.Default); + ((IDisposable)worker).Dispose(); - int initialSegmentCount = await _configElementRepository - .GetValue(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken) - .Map(maybeCount => maybeCount.Match(identity, () => 1)); + _workerChannel.TryWrite(new ReleaseMemory(false)); + }, + TaskScheduler.Default); - await worker.WaitForPlaylistSegments(initialSegmentCount, cancellationToken); - - return await GetMultiVariantPlaylist(request); + Either ready = await SessionStartWait.ForReady( + request.ChannelNumber, + worker, + runTask, + initialSegmentCount, + StartDeadline, + cancellationToken); + return await ready.MapAsync(async _ => await GetMultiVariantPlaylist(request)); } private HlsSessionWorker GetSessionWorker(StartFFmpegSession request, Option targetFramerate) => @@ -136,36 +148,13 @@ public class StartFFmpegSessionHandler : IRequestHandler> Validate(StartFFmpegSession request) => - SessionMustBeInactive(request) - .BindT(_ => FolderMustBeEmpty(request)); - - private async Task> SessionMustBeInactive(StartFFmpegSession request) + private void PrepareTranscodeFolder(string channelNumber) { - var result = Optional(_ffmpegSegmenterService.TryAddWorker(request.ChannelNumber, null)) - .Where(success => success) - .Map(_ => Unit.Default) - .ToValidation(new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request))); - - if (result.IsFail && _ffmpegSegmenterService.TryGetWorker( - request.ChannelNumber, - out IHlsSessionWorker worker)) - { - worker?.Touch(Option.None); - } - - return result; - } - - private Task> FolderMustBeEmpty(StartFFmpegSession request) - { - string folder = Path.Combine(FileSystemLayout.TranscodeFolder, request.ChannelNumber); + string folder = Path.Combine(FileSystemLayout.TranscodeFolder, channelNumber); _logger.LogDebug("Preparing transcode folder {Folder}", folder); _localFileSystem.EnsureFolderExists(folder); _localFileSystem.EmptyFolder(folder); - - return Task.FromResult>(Unit.Default); } private async Task GetMultiVariantPlaylist(StartFFmpegSession request) diff --git a/ErsatzTV.Core.Tests/FFmpeg/FFmpegSegmenterServiceTests.cs b/ErsatzTV.Core.Tests/FFmpeg/FFmpegSegmenterServiceTests.cs new file mode 100644 index 000000000..047320568 --- /dev/null +++ b/ErsatzTV.Core.Tests/FFmpeg/FFmpegSegmenterServiceTests.cs @@ -0,0 +1,149 @@ +using ErsatzTV.Core.FFmpeg; +using ErsatzTV.Core.Interfaces.FFmpeg; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using NUnit.Framework; +using Shouldly; + +namespace ErsatzTV.Core.Tests.FFmpeg; + +[TestFixture] +public class FFmpegSegmenterServiceTests +{ + [SetUp] + public void SetUp() => _service = new FFmpegSegmenterService(NullLogger.Instance); + + private FFmpegSegmenterService _service; + + [Test] + public void TryAddWorker_Should_Fail_For_Second_Worker_On_Same_Channel() + { + IHlsSessionWorker first = Substitute.For(); + IHlsSessionWorker second = Substitute.For(); + + _service.TryAddWorker("1", first).ShouldBeTrue(); + _service.TryAddWorker("1", second).ShouldBeFalse(); + + _service.TryGetWorker("1", out IHlsSessionWorker actual).ShouldBeTrue(); + actual.ShouldBeSameAs(first); + } + + [Test] + public void TryAddWorker_Should_Throw_For_Null_Worker() => + Should.Throw(() => _service.TryAddWorker("1", null)); + + [Test] + public void TryAddWorker_Should_Not_Mark_Channel_Active_When_It_Throws() + { + Should.Throw(() => _service.TryAddWorker("1", null)); + + _service.IsActive("1").ShouldBeFalse(); + } + + [Test] + public void RemoveWorker_Should_Ignore_A_Worker_That_Is_Not_Registered() + { + IHlsSessionWorker current = Substitute.For(); + IHlsSessionWorker stale = Substitute.For(); + + _service.TryAddWorker("1", current).ShouldBeTrue(); + _service.RemoveWorker("1", stale); + + _service.IsActive("1").ShouldBeTrue(); + _service.TryGetWorker("1", out IHlsSessionWorker actual).ShouldBeTrue(); + actual.ShouldBeSameAs(current); + } + + [Test] + public void RemoveWorker_Should_Remove_Its_Own_Worker() + { + IHlsSessionWorker worker = Substitute.For(); + + _service.TryAddWorker("1", worker).ShouldBeTrue(); + _service.RemoveWorker("1", worker); + + _service.IsActive("1").ShouldBeFalse(); + _service.Workers.ShouldBeEmpty(); + } + + [Test] + public void OnWorkersChanged_Should_Only_Fire_For_Effective_Changes() + { + IHlsSessionWorker worker = Substitute.For(); + IHlsSessionWorker other = Substitute.For(); + + var count = 0; + _service.OnWorkersChanged += (_, _) => count++; + + _service.TryAddWorker("1", worker); + count.ShouldBe(1); + + _service.TryAddWorker("1", other); + count.ShouldBe(1); + + _service.RemoveWorker("1", other); + count.ShouldBe(1); + + _service.RemoveWorker("1", worker); + count.ShouldBe(2); + } + + [Test] + public async Task LockForStart_Should_Block_A_Second_Start_On_The_Same_Channel() + { + IDisposable first = await _service.LockForStart("1", CancellationToken.None); + + Task second = _service.LockForStart("1", CancellationToken.None); + (await Task.WhenAny(second, Task.Delay(TimeSpan.FromMilliseconds(250)))).ShouldNotBe(second); + + first.Dispose(); + + IDisposable acquired = await second.WaitAsync(TimeSpan.FromSeconds(5)); + acquired.Dispose(); + } + + [Test] + public async Task LockForStart_Should_Not_Block_A_Start_On_Another_Channel() + { + IDisposable first = await _service.LockForStart("1", CancellationToken.None); + + IDisposable second = await _service.LockForStart("2", CancellationToken.None) + .WaitAsync(TimeSpan.FromSeconds(5)); + + second.Dispose(); + first.Dispose(); + } + + [Test] + public async Task LockForStart_Should_Release_When_The_Releaser_Is_Disposed_Twice() + { + IDisposable first = await _service.LockForStart("1", CancellationToken.None); + first.Dispose(); + first.Dispose(); + + IDisposable second = await _service.LockForStart("1", CancellationToken.None) + .WaitAsync(TimeSpan.FromSeconds(5)); + + // a double dispose must not leave 2 permits behind + Task third = _service.LockForStart("1", CancellationToken.None); + (await Task.WhenAny(third, Task.Delay(TimeSpan.FromMilliseconds(250)))).ShouldNotBe(third); + + second.Dispose(); + (await third.WaitAsync(TimeSpan.FromSeconds(5))).Dispose(); + } + + [Test] + public async Task LockForStart_Should_Throw_When_The_Token_Is_Already_Cancelled() + { + IDisposable first = await _service.LockForStart("1", CancellationToken.None); + + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Should.ThrowAsync(() => _service.LockForStart("1", cts.Token)); + + // the cancelled waiter must not have taken the lock + first.Dispose(); + (await _service.LockForStart("1", CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5))).Dispose(); + } +} diff --git a/ErsatzTV.Core.Tests/FFmpeg/SessionStartWaitTests.cs b/ErsatzTV.Core.Tests/FFmpeg/SessionStartWaitTests.cs new file mode 100644 index 000000000..7a9438f9c --- /dev/null +++ b/ErsatzTV.Core.Tests/FFmpeg/SessionStartWaitTests.cs @@ -0,0 +1,132 @@ +using ErsatzTV.Core.FFmpeg; +using ErsatzTV.Core.Interfaces.FFmpeg; +using NSubstitute; +using NUnit.Framework; +using Shouldly; + +namespace ErsatzTV.Core.Tests.FFmpeg; + +[TestFixture] +public class SessionStartWaitTests +{ + private static readonly TimeSpan ShortDeadline = TimeSpan.FromMilliseconds(250); + + [Test] + public async Task ForReady_Should_Succeed_When_The_Playlist_Becomes_Ready() + { + var wait = new TaskCompletionSource(); + IHlsSessionWorker worker = WorkerWaiting(wait.Task, out _); + + var run = new TaskCompletionSource(); + Task> ready = SessionStartWait.ForReady( + "1", + worker, + run.Task, + 1, + TimeSpan.FromSeconds(30), + CancellationToken.None); + + wait.SetResult(); + + (await ready.WaitAsync(TimeSpan.FromSeconds(5))).IsRight.ShouldBeTrue(); + } + + [Test] + public async Task ForReady_Should_Fail_When_The_Run_Task_Ends_First() + { + IHlsSessionWorker worker = WorkerWaiting(NeverReady(), out _); + + var run = new TaskCompletionSource(); + Task> ready = SessionStartWait.ForReady( + "1", + worker, + run.Task, + 1, + TimeSpan.FromSeconds(30), + CancellationToken.None); + + run.SetResult(); + + Either result = await ready.WaitAsync(TimeSpan.FromSeconds(5)); + ErrorMessage(result).ShouldContain("ended before the playlist was ready"); + } + + [Test] + public async Task ForReady_Should_Stop_The_Wait_When_The_Run_Task_Ends_First() + { + IHlsSessionWorker worker = WorkerWaiting(NeverReady(), out Func waitToken); + + var run = new TaskCompletionSource(); + Task> ready = SessionStartWait.ForReady( + "1", + worker, + run.Task, + 1, + TimeSpan.FromSeconds(30), + CancellationToken.None); + + run.SetResult(); + await ready.WaitAsync(TimeSpan.FromSeconds(5)); + + // an abandoned wait polls on a timer, so it has to be cancelled, not just left behind + waitToken().IsCancellationRequested.ShouldBeTrue(); + } + + [Test] + public async Task ForReady_Should_Fail_At_The_Deadline_When_The_Playlist_Never_Appears() + { + IHlsSessionWorker worker = WorkerWaiting(NeverReady(), out _); + + Either result = await SessionStartWait.ForReady( + "1", + worker, + NeverReady(), + 1, + ShortDeadline, + CancellationToken.None) + .WaitAsync(TimeSpan.FromSeconds(5)); + + ErrorMessage(result).ShouldContain("did not become ready"); + } + + [Test] + public async Task ForReady_Should_Propagate_Cancellation_From_The_Request() + { + IHlsSessionWorker worker = WorkerWaiting(NeverReady(), out _); + + using var cts = new CancellationTokenSource(); + Task> ready = SessionStartWait.ForReady( + "1", + worker, + NeverReady(), + 1, + TimeSpan.FromSeconds(30), + cts.Token); + + await cts.CancelAsync(); + + await Should.ThrowAsync(() => ready.WaitAsync(TimeSpan.FromSeconds(5))); + } + + private static Task NeverReady() => new TaskCompletionSource().Task; + + private static string ErrorMessage(Either result) => + result.Match(_ => string.Empty, error => error.Value); + + // returns a worker whose wait completes with `wait`, plus an accessor for the token it was given + private static IHlsSessionWorker WorkerWaiting(Task wait, out Func waitToken) + { + CancellationToken captured = CancellationToken.None; + waitToken = () => captured; + + IHlsSessionWorker worker = Substitute.For(); + worker.WaitForPlaylistSegments(Arg.Any(), Arg.Any()) + .Returns(callInfo => + { + captured = callInfo.Arg(); + return wait.WaitAsync(captured); + }); + + return worker; + } +} diff --git a/ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs b/ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs index 08dea5023..4d319047f 100644 --- a/ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs +++ b/ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs @@ -6,36 +6,28 @@ namespace ErsatzTV.Core.FFmpeg; public class FFmpegSegmenterService(ILogger logger) : IFFmpegSegmenterService { + private readonly ConcurrentDictionary _startLocks = new(); private readonly ConcurrentDictionary _sessionWorkers = new(); public event EventHandler OnWorkersChanged; public ICollection Workers => _sessionWorkers.Values; + public async Task LockForStart(string channelNumber, CancellationToken cancellationToken) + { + SemaphoreSlim slim = _startLocks.GetOrAdd(channelNumber, _ => new SemaphoreSlim(1, 1)); + await slim.WaitAsync(cancellationToken); + return new StartLockReleaser(slim); + } + public bool TryGetWorker(string channelNumber, out IHlsSessionWorker worker) => _sessionWorkers.TryGetValue(channelNumber, out worker); public bool TryAddWorker(string channelNumber, IHlsSessionWorker worker) { - var result = false; - - // check for worker - if (TryGetWorker(channelNumber, out IHlsSessionWorker existing)) - { - // if worker is null, pretend we added it - if (existing is null) - { - result = true; - } - - // if worker is not null, we cannot add one (so result should stay false) - } - else - { - // worker does not exist, so try adding a null one - result = _sessionWorkers.TryAdd(channelNumber, worker); - } + ArgumentNullException.ThrowIfNull(worker); + bool result = _sessionWorkers.TryAdd(channelNumber, worker); if (result) { OnWorkersChanged?.Invoke(this, EventArgs.Empty); @@ -44,16 +36,12 @@ public class FFmpegSegmenterService(ILogger logger) : IF return result; } - public void AddOrUpdateWorker(string channelNumber, IHlsSessionWorker worker) - { - _sessionWorkers.AddOrUpdate(channelNumber, _ => worker, (_, _) => worker); - OnWorkersChanged?.Invoke(this, EventArgs.Empty); - } - - public void RemoveWorker(string channelNumber, out IHlsSessionWorker inactiveWorker) + public void RemoveWorker(string channelNumber, IHlsSessionWorker worker) { - _sessionWorkers.TryRemove(channelNumber, out inactiveWorker); - OnWorkersChanged?.Invoke(this, EventArgs.Empty); + if (_sessionWorkers.TryRemove(new KeyValuePair(channelNumber, worker))) + { + OnWorkersChanged?.Invoke(this, EventArgs.Empty); + } } public bool IsActive(string channelNumber) => _sessionWorkers.ContainsKey(channelNumber); @@ -62,11 +50,8 @@ public class FFmpegSegmenterService(ILogger logger) : IF { if (_sessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker)) { - if (worker != null) - { - await worker.Cancel(cancellationToken); - return true; - } + await worker.Cancel(cancellationToken); + return true; } return false; @@ -76,7 +61,7 @@ public class FFmpegSegmenterService(ILogger logger) : IF { if (_sessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker)) { - worker?.Touch(fileName); + worker.Touch(fileName); } } @@ -84,14 +69,11 @@ public class FFmpegSegmenterService(ILogger logger) : IF { if (_sessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker)) { - if (worker != null) - { - logger.LogInformation( - "Playout has been updated for channel {ChannelNumber}, HLS segmenter will skip ahead to catch up", - channelNumber); - - worker.PlayoutUpdated(); - } + logger.LogInformation( + "Playout has been updated for channel {ChannelNumber}, HLS segmenter will skip ahead to catch up", + channelNumber); + + worker.PlayoutUpdated(); } } } diff --git a/ErsatzTV.Core/FFmpeg/SessionStartWait.cs b/ErsatzTV.Core/FFmpeg/SessionStartWait.cs new file mode 100644 index 000000000..66b34992e --- /dev/null +++ b/ErsatzTV.Core/FFmpeg/SessionStartWait.cs @@ -0,0 +1,52 @@ +using ErsatzTV.Core.Interfaces.FFmpeg; + +namespace ErsatzTV.Core.FFmpeg; + +public static class SessionStartWait +{ + public static async Task> ForReady( + string channelNumber, + IHlsSessionWorker worker, + Task runTask, + int initialSegmentCount, + TimeSpan startDeadline, + CancellationToken cancellationToken) + { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + timeout.CancelAfter(startDeadline); + + Task waitTask = worker.WaitForPlaylistSegments(initialSegmentCount, timeout.Token); + + try + { + Task first = await Task.WhenAny(waitTask, runTask); + if (first == runTask) + { + return BaseError.New($"Session for channel {channelNumber} ended before the playlist was ready"); + } + + await waitTask; + + return Unit.Default; + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + return BaseError.New($"Session for channel {channelNumber} did not become ready in {startDeadline}"); + } + finally + { + // the wait polls on a timer; disposing the token source does not stop it, so an + // abandoned wait would poll for the life of the process + await timeout.CancelAsync(); + + try + { + await waitTask; + } + catch (Exception) + { + // the start already has its result + } + } + } +} diff --git a/ErsatzTV.Core/FFmpeg/StartLockReleaser.cs b/ErsatzTV.Core/FFmpeg/StartLockReleaser.cs new file mode 100644 index 000000000..713c60c8d --- /dev/null +++ b/ErsatzTV.Core/FFmpeg/StartLockReleaser.cs @@ -0,0 +1,25 @@ +namespace ErsatzTV.Core.FFmpeg; + +public sealed class StartLockReleaser(SemaphoreSlim slim) : IDisposable +{ + private bool _disposedValue; + + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + private void Dispose(bool disposing) + { + if (!_disposedValue) + { + if (disposing) + { + slim.Release(); + } + + _disposedValue = true; + } + } +} diff --git a/ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs b/ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs index 21c3222ca..dcabfec19 100644 --- a/ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs +++ b/ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs @@ -4,10 +4,10 @@ public interface IFFmpegSegmenterService { ICollection Workers { get; } event EventHandler OnWorkersChanged; + Task LockForStart(string channelNumber, CancellationToken cancellationToken); bool TryGetWorker(string channelNumber, out IHlsSessionWorker worker); bool TryAddWorker(string channelNumber, IHlsSessionWorker worker); - void AddOrUpdateWorker(string channelNumber, IHlsSessionWorker worker); - void RemoveWorker(string channelNumber, out IHlsSessionWorker inactiveWorker); + void RemoveWorker(string channelNumber, IHlsSessionWorker worker); bool IsActive(string channelNumber); Task StopChannel(string channelNumber, CancellationToken cancellationToken); void TouchChannel(string channelNumber, string fileName); diff --git a/ErsatzTV/Controllers/IptvController.cs b/ErsatzTV/Controllers/IptvController.cs index 9a9cf52c0..f8f0ae9fb 100644 --- a/ErsatzTV/Controllers/IptvController.cs +++ b/ErsatzTV/Controllers/IptvController.cs @@ -159,7 +159,7 @@ public class IptvController : StreamingControllerBase { // _logger.LogDebug("Checking for session worker for channel {Channel}", channelNumber); - if (_ffmpegSegmenterService.TryGetWorker(channelNumber, out IHlsSessionWorker worker) && worker is not null) + if (_ffmpegSegmenterService.TryGetWorker(channelNumber, out IHlsSessionWorker worker)) { // _logger.LogDebug("Trimming playlist for channel {Channel}", channelNumber); @@ -185,10 +185,13 @@ public class IptvController : StreamingControllerBase [HttpGet("iptv/channel/{channelNumber}.m3u8")] public async Task GetHttpLiveStreamingVideo( string channelNumber, + CancellationToken cancellationToken, [FromQuery] string mode = "mixed") { - Option maybeChannel = await _mediator.Send(new GetChannelByNumber(channelNumber)); + Option maybeChannel = await _mediator.Send( + new GetChannelByNumber(channelNumber), + cancellationToken); if (maybeChannel.IsNone || !await maybeChannel.Map(c => c.IsEnabled).IfNoneAsync(false)) { return NotFound(); @@ -239,7 +242,7 @@ public class IptvController : StreamingControllerBase Request.Host.ToString(), Request.PathBase, AccessTokenQuery()); - Either result = await _mediator.Send(request); + Either result = await _mediator.Send(request, cancellationToken); return result.Match( multiVariantPlaylist => { @@ -276,7 +279,8 @@ public class IptvController : StreamingControllerBase Request.Host.ToString(), channelNumber, mode, - Request.Query["access_token"])) + Request.Query["access_token"]), + cancellationToken) .Map(r => r.Match( playlist => Content(playlist, "application/vnd.apple.mpegurl"), error => BadRequest(error.Value))); diff --git a/ErsatzTV/Startup.cs b/ErsatzTV/Startup.cs index 044852f8c..1a7d76bcc 100644 --- a/ErsatzTV/Startup.cs +++ b/ErsatzTV/Startup.cs @@ -633,6 +633,24 @@ public class Startup "HTTP {RequestMethod} {RequestPath} responded {StatusCode} in {Elapsed:0.00} ms from {UserAgent} at {RemoteIP}"; }); + // must be inside the request logging middleware so an aborted request is not logged as an error + app.Use(async (context, next) => + { + try + { + await next(context); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + // the client is gone, so nothing can be written to the connection; 499 only + // keeps the request log honest about why the request ended + if (!context.Response.HasStarted) + { + context.Response.StatusCode = 499; + } + } + }); + app.UseRequestLocalization(options => { CultureInfo[] cinfo = CultureInfo.GetCultures(CultureTypes.AllCultures & ~CultureTypes.NeutralCultures);