Merge branch 'feat/235-s4-trakt' into feat/235-async-contract

ErsatzTV.Tests/Services/WorkerServiceTests.cs

@@ -0,0 +1,101 @@
+using System.Threading.Channels;
+using ErsatzTV.Application;
+using ErsatzTV.Application.MediaCollections;
+using ErsatzTV.Core;
+using ErsatzTV.Core.Interfaces.Locking;
+using ErsatzTV.Services;
+using LanguageExt;
+using MediatR;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using NUnit.Framework;
+using Shouldly;
+using Unit = LanguageExt.Unit;
+
+namespace ErsatzTV.Tests.Services;
+
+[TestFixture]
+public class WorkerServiceTests
+{
+    // F7 regression (issue #235): the global Trakt lock is acquired by SchedulerService and released
+    // only when the *terminal* (Unlock: true) message of a batch is processed by WorkerService. If the
+    // worker stops before reaching that terminal message (shutdown break / channel completion /
+    // cancellation), the release never fires and the in-memory Trakt lock leaks for the life of the
+    // process. WorkerService must release a held Trakt lock as a compensating action on shutdown.
+    [Test]
+    public async Task Should_Release_Held_Trakt_Lock_When_Worker_Stops_Before_Terminal_Message()
+    {
+        var channel = Channel.CreateUnbounded<IBackgroundServiceRequest>();
+
+        // A Trakt batch: only the terminal message carries Unlock: true. If the worker never processes
+        // it, the handler-side UnlockTrakt() never runs.
+        AddTraktList nonTerminal = AddTraktList.Existing("user", "list-1", false);
+        AddTraktList terminal = AddTraktList.Existing("user", "list-2", true);
+        await channel.Writer.WriteAsync(nonTerminal);
+        await channel.Writer.WriteAsync(terminal);
+
+        // Stateful stand-in for the singleton EntityLocker: the lock starts held (a batch acquired it).
+        var traktLocked = 1;
+        var locker = Substitute.For<IEntityLocker>();
+        locker.IsTraktLocked().Returns(_ => Volatile.Read(ref traktLocked) == 1);
+        locker.UnlockTrakt().Returns(_ => Interlocked.Exchange(ref traktLocked, 0) == 1);
+
+        // Gate: processing the first (non-terminal) message parks on the stopping token, guaranteeing
+        // the worker never reaches the terminal (Unlock: true) message before it is stopped.
+        var firstSeen = new TaskCompletionSource();
+        var processed = new List<AddTraktList>();
+        var mediator = Substitute.For<IMediator>();
+        mediator.Send(Arg.Any<AddTraktList>(), Arg.Any<CancellationToken>())
+            .Returns(async call =>
+            {
+                lock (processed)
+                {
+                    processed.Add(call.Arg<AddTraktList>());
+                }
+
+                firstSeen.TrySetResult();
+
+                // Block on the stopping token so the loop parks here until StopAsync cancels it.
+                await Task.Delay(Timeout.Infinite, call.Arg<CancellationToken>());
+                return (Either<BaseError, Unit>)Unit.Default;
+            });
+
+        var provider = Substitute.For<IServiceProvider>();
+        provider.GetService(typeof(IMediator)).Returns(mediator);
+        var scope = Substitute.For<IServiceScope>();
+        scope.ServiceProvider.Returns(provider);
+        var scopeFactory = Substitute.For<IServiceScopeFactory>();
+        scopeFactory.CreateScope().Returns(scope);
+
+        var worker = new WorkerService(
+            channel.Reader,
+            scopeFactory,
+            locker,
+            NullLogger<WorkerService>.Instance);
+
+        await worker.StartAsync(CancellationToken.None);
+
+        // Wait until the first message is actively being processed (parked on the stopping token).
+        await firstSeen.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        // Shut the worker down: cancels the stopping token -> parked Delay throws -> loop breaks
+        // before the terminal message is ever processed.
+        await worker.StopAsync(CancellationToken.None);
+
+        // The compensating release must have fired even though the terminal message was never handled.
+        locker.IsTraktLocked().ShouldBeFalse();
+        locker.Received(1).UnlockTrakt();
+
+        // Prove the leak scenario is genuine: we stopped after the non-terminal message but before the
+        // terminal (Unlock: true) one, so the handler-side release could not have run.
+        List<AddTraktList> seen;
+        lock (processed)
+        {
+            seen = processed.ToList();
+        }
+
+        seen.ShouldContain(nonTerminal);
+        seen.ShouldNotContain(terminal);
+    }
+}

ErsatzTV/Services/WorkerService.cs

@@ -9,6 +9,7 @@ using ErsatzTV.Application.MediaCollections;
 using ErsatzTV.Application.Playouts;
 using ErsatzTV.Application.Subtitles;
 using ErsatzTV.Core;
+using ErsatzTV.Core.Interfaces.Locking;
 using MediatR;
 
 namespace ErsatzTV.Services;
@@ -16,16 +17,19 @@ namespace ErsatzTV.Services;
 public class WorkerService : BackgroundService
 {
     private readonly ChannelReader<IBackgroundServiceRequest> _channel;
+    private readonly IEntityLocker _entityLocker;
     private readonly ILogger<WorkerService> _logger;
     private readonly IServiceScopeFactory _serviceScopeFactory;
 
     public WorkerService(
         ChannelReader<IBackgroundServiceRequest> channel,
         IServiceScopeFactory serviceScopeFactory,
+        IEntityLocker entityLocker,
         ILogger<WorkerService> logger)
     {
         _channel = channel;
         _serviceScopeFactory = serviceScopeFactory;
+        _entityLocker = entityLocker;
         _logger = logger;
     }
 
@@ -143,5 +147,19 @@ public class WorkerService : BackgroundService
         {
             _logger.LogInformation("Worker service shutting down");
         }
+        finally
+        {
+            // The global Trakt lock is acquired by SchedulerService/TraktController and released only
+            // when the *terminal* (Unlock: true) message of a batch is processed here. If this loop
+            // stops before reaching that message - shutdown break above, channel completion, or the
+            // reader throwing on cancellation - the release never fires and the in-memory Trakt lock
+            // leaks for the remaining life of the process (subsequent Trakt operations 409 forever).
+            // Make the batch-release loss-tolerant with a compensating release on worker shutdown.
+            if (_entityLocker.IsTraktLocked())
+            {
+                _logger.LogDebug("Releasing held Trakt lock during worker shutdown");
+                _entityLocker.UnlockTrakt();
+            }
+        }
     }
 }