fix(235-F7): release leaked Trakt lock on worker shutdown

The global Trakt lock is acquired by SchedulerService.RefreshTraktLists /
MatchTraktLists (and TraktController) and released only when the *terminal*
message of a batch — the one carrying Unlock: true (list == traktLists.Last())
— is processed by WorkerService, whose handler (AddTraktListHandler /
MatchTraktListItemsHandler) calls IEntityLocker.UnlockTrakt() in a finally.

WorkerService.ExecuteAsync breaks out of the read loop on
stoppingToken.IsCancellationRequested (and exits on channel completion /
reader cancellation) BEFORE processing the next message. If shutdown lands
after a batch is enqueued but before its terminal Unlock: true message is
handled, UnlockTrakt() never runs and the in-memory Trakt lock leaks for the
rest of the process lifetime (subsequent Trakt operations 409 forever).

Fix (option a): make the batch-release loss-tolerant with a compensating
release in a finally around the read loop — if the Trakt lock is still held
when the worker stops, release it. Chosen over tracking pending ownership
(b) because the lock is a global singleton and WorkerService is its sole
batch-release site, so "held at shutdown" unambiguously means "the terminal
release was lost"; covers all three exit paths (break / channel completion /
cancellation) in one place. Same lock-lifecycle class as #231/#233/#234.

Regression test: WorkerServiceTests gates the first (non-terminal) batch
message on the stopping token, then StopAsync-cancels so the worker breaks
before the terminal Unlock: true message — asserts the lock is released and
the terminal message was never processed. Proven non-vacuous: inverting the
finally condition fails the test.

Backend-only; no controller/DTO/SPA/OpenAPI impact.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-11 17:57:41 +02:00
co-authored by Claude Opus 4.8
parent 8e2f995492
commit d02f953922
2 changed files with 119 additions and 0 deletions
@@ -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);
}
}
+18
View File
@@ -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();
}
}
}
}