fix: channel startup (#3002)

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
+  - Symptoms included:
-  - 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
+      - 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
   - `<country>` 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

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<NextSessionWorker> sessionWorkerLogger)
     : NextChannelHandlerBase(fileSystem), IRequestHandler<StartFFmpegNextSession, Either<BaseError, string>>
 {
+    private static readonly TimeSpan StartDeadline = TimeSpan.FromSeconds(30);
+
     private readonly IFileSystem _fileSystem = fileSystem;
 
-    public Task<Either<BaseError, string>> Handle(
+    public async Task<Either<BaseError, string>> 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<BaseError, Task<string>, string>(identity));
-#pragma warning restore VSTHRD103
-
-    private async Task<string> StartProcess(
         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<string>.None);
+            return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request));
+        }
+
+        Validation<BaseError, string> maybeChannelBinary = await ChannelBinaryMustExist();
+        if (maybeChannelBinary.IsFail)
+        {
+            return maybeChannelBinary.FailToSeq().Head();
+        }
+
+        string channelBinary = maybeChannelBinary.SuccessToSeq().Head();
+
         Option<TimeSpan> idleTimeout = Option<TimeSpan>.None;
 
         // Option<FrameRate> targetFramerate = await mediator.Send(
         //     new GetChannelFramerate(request.ChannelNumber),
         //     cancellationToken);
 
+        int initialSegmentCount = await configElementRepository
+            .GetValue<int>(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken)
+            .Map(maybeCount => maybeCount.Match(identity, () => 1));
+
+        Option<ChannelViewModel> 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<FFmpegProfileViewModel> 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<int>(ConfigElementKey.FFmpegSegmenterTimeout, cancellationToken)
@@ -67,111 +99,53 @@ public class StartFFmpegNextSessionHandler(
 
         await mediator.Send(new RefreshGraphicsElements(), cancellationToken);
 
-        ChannelConfig config = await channelConfigConverter.ToNext(
+        PrepareTranscodeFolder(request.ChannelNumber);
-            validationResult.Channel,
+
-            validationResult.FfmpegProfile,
+        ChannelConfig config = await channelConfigConverter.ToNext(channel, ffmpegProfile, cancellationToken);
-            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)
+        Task runTask = worker.Run(request.ChannelNumber, idleTimeout, hostApplicationLifetime.ApplicationStopping);
-            .ContinueWith(
+        _ = 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
+        Either<BaseError, Unit> ready = await SessionStartWait.ForReady(
-            .GetValue<int>(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken)
+            request.ChannelNumber,
-            .Map(maybeCount => maybeCount.Match(identity, () => 1));
+            worker,
-
+            runTask,
-        await worker.WaitForPlaylistSegments(initialSegmentCount, cancellationToken);
+            initialSegmentCount,
-
+            StartDeadline,
-        return await GetMultiVariantPlaylist(request);
+            cancellationToken);
-    }
+        return await ready.MapAsync(async _ => await GetMultiVariantPlaylist(request));
-
-    private Task<Validation<BaseError, ValidationResult>> 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<Validation<BaseError, Unit>> SessionMustBeInactive(StartFFmpegNextSession request)
-    {
-        var result = Optional(ffmpegSegmenterService.TryAddWorker(request.ChannelNumber, null))
-            .Where(success => success)
-            .Map(_ => Unit.Default)
-            .ToValidation<BaseError>(new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request)));
-
-        if (result.IsFail && ffmpegSegmenterService.TryGetWorker(
-                request.ChannelNumber,
-                out IHlsSessionWorker worker))
-        {
-            worker?.Touch(Option<string>.None);
-        }
-
-        return result;
     }
 
-    private Task<Validation<BaseError, Unit>> 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<Validation<BaseError, Unit>>(Unit.Default);
-    }
-
-    private async Task<Validation<BaseError, ValidationResult>> ChannelMustExist(
-        StartFFmpegNextSession request,
-        ValidationResult result,
-        CancellationToken cancellationToken)
-    {
-        Option<ChannelViewModel> 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<Validation<BaseError, ValidationResult>> FFmpegProfileMustExist(
-        ValidationResult result,
-        CancellationToken cancellationToken)
-    {
-        Option<FFmpegProfileViewModel> 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<string> 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);
 }

ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs

@@ -22,6 +22,8 @@ namespace ErsatzTV.Application.Streaming;
 
 public class StartFFmpegSessionHandler : IRequestHandler<StartFFmpegSession, Either<BaseError, string>>
 {
+    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<StartFFmpegSession, Eit
         _workerChannel = workerChannel;
     }
 
-    public Task<Either<BaseError, string>> Handle(StartFFmpegSession request, CancellationToken cancellationToken) =>
+    public async Task<Either<BaseError, string>> 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<BaseError, Task<string>, string>(identity));
-#pragma warning restore VSTHRD103
-
-    private async Task<string> StartProcess(StartFFmpegSession request, CancellationToken cancellationToken)
     {
+        using IDisposable releaser =
+            await _ffmpegSegmenterService.LockForStart(request.ChannelNumber, cancellationToken);
+
+        if (_ffmpegSegmenterService.TryGetWorker(request.ChannelNumber, out IHlsSessionWorker existing))
+        {
+            existing.Touch(Option<string>.None);
+            return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request));
+        }
+
         Option<TimeSpan> idleTimeout = await _configElementRepository
             .GetValue<int>(ConfigElementKey.FFmpegSegmenterTimeout, cancellationToken)
             .Map(maybeTimeout => maybeTimeout.Match(i => TimeSpan.FromSeconds(i), () => TimeSpan.FromMinutes(1)));
@@ -84,6 +87,10 @@ public class StartFFmpegSessionHandler : IRequestHandler<StartFFmpegSession, Eit
             new GetChannelFramerate(request.ChannelNumber),
             cancellationToken);
 
+        int initialSegmentCount = await _configElementRepository
+            .GetValue<int>(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken)
+            .Map(maybeCount => maybeCount.Match(identity, () => 1));
+
         // disable idle timeout when configured to keep running
         Option<ChannelViewModel> channel =
             await _mediator.Send(new GetChannelByNumber(request.ChannelNumber), cancellationToken);
@@ -94,30 +101,35 @@ public class StartFFmpegSessionHandler : IRequestHandler<StartFFmpegSession, Eit
 
         await _mediator.Send(new RefreshGraphicsElements(), cancellationToken);
 
-        HlsSessionWorker worker = GetSessionWorker(request, targetFramerate);
+        PrepareTranscodeFolder(request.ChannelNumber);
 
-        _ffmpegSegmenterService.AddOrUpdateWorker(request.ChannelNumber, worker);
+        HlsSessionWorker worker = GetSessionWorker(request, targetFramerate);
+        if (!_ffmpegSegmenterService.TryAddWorker(request.ChannelNumber, worker))
+        {
+            return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request));
+        }
 
         // fire and forget worker
-        _ = worker.Run(request.ChannelNumber, idleTimeout, _hostApplicationLifetime.ApplicationStopping)
+        Task runTask = worker.Run(request.ChannelNumber, idleTimeout, _hostApplicationLifetime.ApplicationStopping);
-            .ContinueWith(
+        _ = runTask.ContinueWith(
-                _ =>
+            _ =>
-                {
+            {
-                    _ffmpegSegmenterService.RemoveWorker(request.ChannelNumber, out IHlsSessionWorker inactiveWorker);
+                _ffmpegSegmenterService.RemoveWorker(request.ChannelNumber, worker);
-
-                    inactiveWorker?.Dispose();
 
-                    _workerChannel.TryWrite(new ReleaseMemory(false));
+                ((IDisposable)worker).Dispose();
-                },
-                TaskScheduler.Default);
 
-        int initialSegmentCount = await _configElementRepository
+                _workerChannel.TryWrite(new ReleaseMemory(false));
-            .GetValue<int>(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken)
+            },
-            .Map(maybeCount => maybeCount.Match(identity, () => 1));
+            TaskScheduler.Default);
 
-        await worker.WaitForPlaylistSegments(initialSegmentCount, cancellationToken);
+        Either<BaseError, Unit> ready = await SessionStartWait.ForReady(
-
+            request.ChannelNumber,
-        return await GetMultiVariantPlaylist(request);
+            worker,
+            runTask,
+            initialSegmentCount,
+            StartDeadline,
+            cancellationToken);
+        return await ready.MapAsync(async _ => await GetMultiVariantPlaylist(request));
     }
 
     private HlsSessionWorker GetSessionWorker(StartFFmpegSession request, Option<FrameRate> targetFramerate) =>
@@ -136,36 +148,13 @@ public class StartFFmpegSessionHandler : IRequestHandler<StartFFmpegSession, Eit
                 targetFramerate)
         };
 
-    private Task<Validation<BaseError, Unit>> Validate(StartFFmpegSession request) =>
+    private void PrepareTranscodeFolder(string channelNumber)
-        SessionMustBeInactive(request)
-            .BindT(_ => FolderMustBeEmpty(request));
-
-    private async Task<Validation<BaseError, Unit>> SessionMustBeInactive(StartFFmpegSession request)
     {
-        var result = Optional(_ffmpegSegmenterService.TryAddWorker(request.ChannelNumber, null))
+        string folder = Path.Combine(FileSystemLayout.TranscodeFolder, channelNumber);
-            .Where(success => success)
-            .Map(_ => Unit.Default)
-            .ToValidation<BaseError>(new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request)));
-
-        if (result.IsFail && _ffmpegSegmenterService.TryGetWorker(
-                request.ChannelNumber,
-                out IHlsSessionWorker worker))
-        {
-            worker?.Touch(Option<string>.None);
-        }
-
-        return result;
-    }
-
-    private Task<Validation<BaseError, Unit>> FolderMustBeEmpty(StartFFmpegSession request)
-    {
-        string folder = Path.Combine(FileSystemLayout.TranscodeFolder, request.ChannelNumber);
         _logger.LogDebug("Preparing transcode folder {Folder}", folder);
 
         _localFileSystem.EnsureFolderExists(folder);
         _localFileSystem.EmptyFolder(folder);
-
-        return Task.FromResult<Validation<BaseError, Unit>>(Unit.Default);
     }
 
     private async Task<string> GetMultiVariantPlaylist(StartFFmpegSession request)

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<FFmpegSegmenterService>.Instance);
+
+    private FFmpegSegmenterService _service;
+
+    [Test]
+    public void TryAddWorker_Should_Fail_For_Second_Worker_On_Same_Channel()
+    {
+        IHlsSessionWorker first = Substitute.For<IHlsSessionWorker>();
+        IHlsSessionWorker second = Substitute.For<IHlsSessionWorker>();
+
+        _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<ArgumentNullException>(() => _service.TryAddWorker("1", null));
+
+    [Test]
+    public void TryAddWorker_Should_Not_Mark_Channel_Active_When_It_Throws()
+    {
+        Should.Throw<ArgumentNullException>(() => _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>();
+        IHlsSessionWorker stale = Substitute.For<IHlsSessionWorker>();
+
+        _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<IHlsSessionWorker>();
+
+        _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>();
+        IHlsSessionWorker other = Substitute.For<IHlsSessionWorker>();
+
+        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<IDisposable> 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<IDisposable> 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<OperationCanceledException>(() => _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();
+    }
+}

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<Either<BaseError, Unit>> 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<Either<BaseError, Unit>> ready = SessionStartWait.ForReady(
+            "1",
+            worker,
+            run.Task,
+            1,
+            TimeSpan.FromSeconds(30),
+            CancellationToken.None);
+
+        run.SetResult();
+
+        Either<BaseError, Unit> 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<CancellationToken> waitToken);
+
+        var run = new TaskCompletionSource();
+        Task<Either<BaseError, Unit>> 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<BaseError, Unit> 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<Either<BaseError, Unit>> ready = SessionStartWait.ForReady(
+            "1",
+            worker,
+            NeverReady(),
+            1,
+            TimeSpan.FromSeconds(30),
+            cts.Token);
+
+        await cts.CancelAsync();
+
+        await Should.ThrowAsync<OperationCanceledException>(() => ready.WaitAsync(TimeSpan.FromSeconds(5)));
+    }
+
+    private static Task NeverReady() => new TaskCompletionSource().Task;
+
+    private static string ErrorMessage(Either<BaseError, Unit> 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<CancellationToken> waitToken)
+    {
+        CancellationToken captured = CancellationToken.None;
+        waitToken = () => captured;
+
+        IHlsSessionWorker worker = Substitute.For<IHlsSessionWorker>();
+        worker.WaitForPlaylistSegments(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(callInfo =>
+            {
+                captured = callInfo.Arg<CancellationToken>();
+                return wait.WaitAsync(captured);
+            });
+
+        return worker;
+    }
+}

ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs

@@ -6,36 +6,28 @@ namespace ErsatzTV.Core.FFmpeg;
 
 public class FFmpegSegmenterService(ILogger<FFmpegSegmenterService> logger) : IFFmpegSegmenterService
 {
+    private readonly ConcurrentDictionary<string, SemaphoreSlim> _startLocks = new();
     private readonly ConcurrentDictionary<string, IHlsSessionWorker> _sessionWorkers = new();
 
     public event EventHandler OnWorkersChanged;
 
     public ICollection<IHlsSessionWorker> Workers => _sessionWorkers.Values;
 
+    public async Task<IDisposable> 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;
+        ArgumentNullException.ThrowIfNull(worker);
-
-        // 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);
-        }
 
+        bool result = _sessionWorkers.TryAdd(channelNumber, worker);
         if (result)
         {
             OnWorkersChanged?.Invoke(this, EventArgs.Empty);
@@ -44,16 +36,12 @@ public class FFmpegSegmenterService(ILogger<FFmpegSegmenterService> logger) : IF
         return result;
     }
 
-    public void AddOrUpdateWorker(string channelNumber, IHlsSessionWorker worker)
+    public void RemoveWorker(string channelNumber, IHlsSessionWorker worker)
-    {
-        _sessionWorkers.AddOrUpdate(channelNumber, _ => worker, (_, _) => worker);
-        OnWorkersChanged?.Invoke(this, EventArgs.Empty);
-    }
-
-    public void RemoveWorker(string channelNumber, out IHlsSessionWorker inactiveWorker)
     {
-        _sessionWorkers.TryRemove(channelNumber, out inactiveWorker);
+        if (_sessionWorkers.TryRemove(new KeyValuePair<string, IHlsSessionWorker>(channelNumber, worker)))
-        OnWorkersChanged?.Invoke(this, EventArgs.Empty);
+        {
+            OnWorkersChanged?.Invoke(this, EventArgs.Empty);
+        }
     }
 
     public bool IsActive(string channelNumber) => _sessionWorkers.ContainsKey(channelNumber);
@@ -62,11 +50,8 @@ public class FFmpegSegmenterService(ILogger<FFmpegSegmenterService> 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<FFmpegSegmenterService> logger) : IF
     {
         if (_sessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker))
         {
-            worker?.Touch(fileName);
+            worker.Touch(fileName);
         }
     }
 
@@ -84,14 +69,11 @@ public class FFmpegSegmenterService(ILogger<FFmpegSegmenterService> 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",
-                logger.LogInformation(
+                channelNumber);
-                    "Playout has been updated for channel {ChannelNumber}, HLS segmenter will skip ahead to catch up",
+
-                    channelNumber);
+            worker.PlayoutUpdated();
-
-                worker.PlayoutUpdated();
-            }
         }
     }
 }

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<Either<BaseError, Unit>> 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
+            }
+        }
+    }
+}

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;
+        }
+    }
+}

ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs

@@ -4,10 +4,10 @@ public interface IFFmpegSegmenterService
 {
     ICollection<IHlsSessionWorker> Workers { get; }
     event EventHandler OnWorkersChanged;
+    Task<IDisposable> 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, IHlsSessionWorker worker);
-    void RemoveWorker(string channelNumber, out IHlsSessionWorker inactiveWorker);
     bool IsActive(string channelNumber);
     Task<bool> StopChannel(string channelNumber, CancellationToken cancellationToken);
     void TouchChannel(string channelNumber, string fileName);

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<IActionResult> GetHttpLiveStreamingVideo(
         string channelNumber,
+        CancellationToken cancellationToken,
         [FromQuery]
         string mode = "mixed")
     {
-        Option<ChannelViewModel> maybeChannel = await _mediator.Send(new GetChannelByNumber(channelNumber));
+        Option<ChannelViewModel> 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<BaseError, string> result = await _mediator.Send(request);
+                Either<BaseError, string> result = await _mediator.Send(request, cancellationToken);
                 return result.Match<IActionResult>(
                     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<IActionResult>(
                         playlist => Content(playlist, "application/vnd.apple.mpegurl"),
                         error => BadRequest(error.Value)));

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);