From 4b74aa23f32f03de88af8ad2b083ded417ee4d8f Mon Sep 17 00:00:00 2001 From: Flaminel Date: Sun, 16 Aug 2026 01:02:21 +0300 Subject: [PATCH] Reduce the number of arr requests made while monitoring searches (#722) --- .../Features/Arr/ArrClientTests.cs | 57 +++++ .../Jobs/SeekerCommandMonitorTests.cs | 238 ++++++++++++++++-- .../Features/Arr/ArrClient.cs | 17 ++ .../Features/Arr/Interfaces/IArrClient.cs | 5 + .../Features/Jobs/SeekerCommandMonitor.cs | 218 ++++++++++++---- 5 files changed, 457 insertions(+), 78 deletions(-) diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Arr/ArrClientTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Arr/ArrClientTests.cs index f90b7179..a6b35d75 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Arr/ArrClientTests.cs +++ b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Arr/ArrClientTests.cs @@ -380,6 +380,63 @@ public class ArrClientTests #endregion + #region GetCommandsAsync + + [Fact] + public async Task GetCommandsAsync_SendsGetWithApiKeyAndExpectedUri() + { + // Arrange + _httpMessageHandler.SetupResponse((_, _) => Task.FromResult(JsonResponse(new[] + { + new ArrCommandStatus(1, ArrCommandState.Started, null), + new ArrCommandStatus(2, ArrCommandState.Completed, "done"), + }))); + + // Act + List commands = await _client.GetCommandsAsync(_arrInstance); + + // Assert + commands.Count.ShouldBe(2); + commands[0].Id.ShouldBe(1); + commands[0].Status.ShouldBe(ArrCommandState.Started); + commands[1].Id.ShouldBe(2); + commands[1].Status.ShouldBe(ArrCommandState.Completed); + commands[1].Message.ShouldBe("done"); + + HttpRequestMessage request = _httpMessageHandler.CapturedRequests.ShouldHaveSingleItem(); + request.Method.ShouldBe(HttpMethod.Get); + request.RequestUri!.AbsolutePath.ShouldBe("/api/v3/command"); + request.Headers.GetValues("x-api-key").ShouldHaveSingleItem().ShouldBe("secret-key"); + } + + [Fact] + public async Task GetCommandsAsync_NullResponseBody_ReturnsEmptyList() + { + // Arrange — body deserializes to null + _httpMessageHandler.SetupResponse((_, _) => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("null", Encoding.UTF8, "application/json"), + })); + + // Act + List commands = await _client.GetCommandsAsync(_arrInstance); + + // Assert + commands.ShouldBeEmpty(); + } + + [Fact] + public async Task GetCommandsAsync_NonSuccessStatus_Throws() + { + // Arrange + _httpMessageHandler.SetupResponse(HttpStatusCode.InternalServerError); + + // Act / Assert + await Should.ThrowAsync(() => _client.GetCommandsAsync(_arrInstance)); + } + + #endregion + #region ShouldRemoveFromQueue [Fact] diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/SeekerCommandMonitorTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/SeekerCommandMonitorTests.cs index 5917b214..3662a29c 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/SeekerCommandMonitorTests.cs +++ b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/SeekerCommandMonitorTests.cs @@ -3,6 +3,7 @@ using Cleanuparr.Domain.Entities.Arr; using Cleanuparr.Domain.Entities.Arr.Queue; using Cleanuparr.Domain.Enums; using Cleanuparr.Infrastructure.Events.Interfaces; +using Cleanuparr.Infrastructure.Features.Arr; using Cleanuparr.Infrastructure.Features.Arr.Interfaces; using Cleanuparr.Infrastructure.Features.Jobs; using Cleanuparr.Infrastructure.Tests.Features.Jobs.TestHelpers; @@ -46,6 +47,8 @@ public class SeekerCommandMonitorTests : IAsyncDisposable serviceProvider.GetService(typeof(DataContext)).Returns(_dataContext); serviceProvider.GetService(typeof(EventsContext)).Returns(_eventsContext); serviceProvider.GetService(typeof(IArrClientFactory)).Returns(arrClientFactory); + serviceProvider.GetService(typeof(IArrQueueIterator)) + .Returns(new ArrQueueIterator(Substitute.For>())); serviceProvider.GetService(typeof(IEventPublisher)).Returns(_eventPublisher); var scope = Substitute.For(); @@ -92,8 +95,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null)); + StubCommandState(ArrCommandState.Completed); // 3 episodes from same season pack share the same DownloadId _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) @@ -148,8 +150,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null)); + StubCommandState(ArrCommandState.Completed); // Queue has records with empty DownloadId and one valid record _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) @@ -202,8 +203,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null)); + StubCommandState(ArrCommandState.Completed); // Two different downloads for the same movie _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) @@ -249,13 +249,12 @@ public class SeekerCommandMonitorTests : IAsyncDisposable ItemTitle = "Stuck Movie", SeasonNumber = 0, Status = SearchCommandStatus.Pending, - CreatedAt = _timeProvider.GetUtcNow().UtcDateTime - TimeSpan.FromMinutes(11) + CreatedAt = _timeProvider.GetUtcNow().UtcDateTime - TimeSpan.FromMinutes(31) }); await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(7, ArrCommandState.Started, null)); + StubCommandState(ArrCommandState.Started, commandId: 7); Task publishTask = CaptureNextPublishedStatus(); @@ -284,8 +283,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(1, ArrCommandState.Failed, null)); + StubCommandState(ArrCommandState.Failed); Task publishTask = CaptureNextPublishedStatus(); @@ -312,8 +310,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(1, state, null)); + StubCommandState(state); Task publishTask = CaptureNextPublishedStatus(); @@ -336,6 +333,8 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); + _arrClient.GetCommandsAsync(Arg.Any()) + .ThrowsAsync(new HttpRequestException("Not found", null, HttpStatusCode.NotFound)); _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) .ThrowsAsync(new HttpRequestException("Not found", null, HttpStatusCode.NotFound)); _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) @@ -362,8 +361,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(1, ArrCommandState.Unknown, null)); + StubCommandState(ArrCommandState.Unknown); Task secondPoll = CaptureNthPoll(2); CaptureNextPublishedStatus(); @@ -389,8 +387,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null)); + StubCommandState(ArrCommandState.Completed); _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) .Returns(new QueueListResponse()); @@ -414,8 +411,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); - _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) - .Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null)); + StubCommandState(ArrCommandState.Completed); _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) .Returns(new QueueListResponse()); @@ -460,7 +456,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable { // Arrange ArrInstance radarrInstance = TestDataContextFactory.AddRadarrInstance(_dataContext); - AddTracker(radarrInstance.Id, Guid.NewGuid(), status: SearchCommandStatus.Completed, age: TimeSpan.FromMinutes(31)); + AddTracker(radarrInstance.Id, Guid.NewGuid(), status: SearchCommandStatus.Completed, age: TimeSpan.FromMinutes(91)); await _dataContext.SaveChangesAsync(); await _eventsContext.SaveChangesAsync(); @@ -510,6 +506,97 @@ public class SeekerCommandMonitorTests : IAsyncDisposable AddTracker(Guid.NewGuid(), eventId); await _eventsContext.SaveChangesAsync(); + Task pendingPublishTask = CaptureNextPublishedStatus(); + + // Act + await _sut.StartAsync(_cts.Token); + _timeProvider.Advance(TimeSpan.FromSeconds(11)); + SearchCommandStatus pendingPublishedStatus = await pendingPublishTask.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert + pendingPublishedStatus.ShouldBe(SearchCommandStatus.Failed); + + await _eventPublisher.Received(1).PublishSearchCompleted( + eventId, SearchCommandStatus.Failed, Arg.Any(), Arg.Any(), Arg.Any?>()); + + await _arrClient.DidNotReceive().GetCommandsAsync(Arg.Any()); + + await WaitForTrackerCountAsync(0); + } + + [Fact] + public async Task Polls_every_command_of_an_instance_in_a_single_request() + { + // Arrange + ArrInstance radarrInstance = TestDataContextFactory.AddRadarrInstance(_dataContext); + AddTracker(radarrInstance.Id, Guid.NewGuid(), commandId: 1); + AddTracker(radarrInstance.Id, Guid.NewGuid(), commandId: 2); + AddTracker(radarrInstance.Id, Guid.NewGuid(), commandId: 3); + await _dataContext.SaveChangesAsync(); + await _eventsContext.SaveChangesAsync(); + + _arrClient.GetCommandsAsync(Arg.Any()) + .Returns( + [ + new ArrCommandStatus(1, ArrCommandState.Completed, null), + new ArrCommandStatus(2, ArrCommandState.Completed, null), + new ArrCommandStatus(3, ArrCommandState.Completed, null), + ]); + _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) + .Returns(new QueueListResponse()); + + CaptureNextPublishedStatus(); + + // Act + await _sut.StartAsync(_cts.Token); + _timeProvider.Advance(TimeSpan.FromSeconds(11)); + await WaitForTrackerCountAsync(0); + + // Assert + await _arrClient.Received(1).GetCommandsAsync(Arg.Any()); + await _arrClient.DidNotReceive().GetCommandStatusAsync(Arg.Any(), Arg.Any()); + await _arrClient.Received(1).GetQueueItemsAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task Completes_a_command_that_is_missing_from_the_command_list() + { + // Arrange + ArrInstance radarrInstance = TestDataContextFactory.AddRadarrInstance(_dataContext); + AddTracker(radarrInstance.Id, Guid.NewGuid(), commandId: 42); + await _dataContext.SaveChangesAsync(); + await _eventsContext.SaveChangesAsync(); + + _arrClient.GetCommandsAsync(Arg.Any()) + .Returns([new ArrCommandStatus(7, ArrCommandState.Started, null)]); + _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) + .Returns(new QueueListResponse()); + + Task publishTask = CaptureNextPublishedStatus(); + + // Act + await _sut.StartAsync(_cts.Token); + _timeProvider.Advance(TimeSpan.FromSeconds(11)); + SearchCommandStatus publishedStatus = await publishTask.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert + publishedStatus.ShouldBe(SearchCommandStatus.Completed); + } + + [Fact] + public async Task Falls_back_to_individual_command_checks_when_the_command_list_fails() + { + // Arrange + ArrInstance radarrInstance = TestDataContextFactory.AddRadarrInstance(_dataContext); + AddTracker(radarrInstance.Id, Guid.NewGuid()); + await _dataContext.SaveChangesAsync(); + await _eventsContext.SaveChangesAsync(); + + _arrClient.GetCommandsAsync(Arg.Any()) + .ThrowsAsync(new HttpRequestException("boom")); + _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) + .Returns(new ArrCommandStatus(1, ArrCommandState.Failed, null)); + Task publishTask = CaptureNextPublishedStatus(); // Act @@ -519,15 +606,116 @@ public class SeekerCommandMonitorTests : IAsyncDisposable // Assert publishedStatus.ShouldBe(SearchCommandStatus.Failed); + await _arrClient.Received(1).GetCommandStatusAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task Reports_grabbed_items_found_beyond_the_first_queue_page() + { + // Arrange + ArrInstance radarrInstance = TestDataContextFactory.AddRadarrInstance(_dataContext); + AddTracker(radarrInstance.Id, Guid.NewGuid()); + await _dataContext.SaveChangesAsync(); + await _eventsContext.SaveChangesAsync(); + + StubCommandState(ArrCommandState.Completed); + + _arrClient.GetQueueItemsAsync(Arg.Any(), 1) + .Returns(new QueueListResponse + { + TotalRecords = 2, + Records = [new QueueRecord { Id = 1, MovieId = 999, Title = "Other.Movie", DownloadId = "OTHER", Protocol = "torrent", Status = "downloading" }] + }); + _arrClient.GetQueueItemsAsync(Arg.Any(), 2) + .Returns(new QueueListResponse + { + TotalRecords = 2, + Records = [new QueueRecord { Id = 2, MovieId = 300, Title = "Wanted.Movie", DownloadId = "WANTED", Protocol = "torrent", Status = "downloading" }] + }); + + TaskCompletionSource?> grabbedTcs = new(); + _eventPublisher.PublishSearchCompleted( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any?>()) + .Returns(Task.CompletedTask) + .AndDoes(ci => grabbedTcs.TrySetResult(ci.ArgAt?>(4))); + + // Act + await _sut.StartAsync(_cts.Token); + _timeProvider.Advance(TimeSpan.FromSeconds(11)); + List? grabbedItems = await grabbedTcs.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert + grabbedItems.ShouldNotBeNull(); + grabbedItems!.ShouldBe(["Wanted.Movie"]); + } + + [Fact] + public async Task Publishes_completed_status_when_the_queue_cannot_be_inspected() + { + // Arrange + ArrInstance radarrInstance = TestDataContextFactory.AddRadarrInstance(_dataContext); + Guid eventId = Guid.NewGuid(); + AddTracker(radarrInstance.Id, eventId); + await _dataContext.SaveChangesAsync(); + await _eventsContext.SaveChangesAsync(); + + StubCommandState(ArrCommandState.Completed); + _arrClient.GetQueueItemsAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new HttpRequestException("queue unavailable")); + + Task publishTask = CaptureNextPublishedStatus(); + + // Act + await _sut.StartAsync(_cts.Token); + _timeProvider.Advance(TimeSpan.FromSeconds(11)); + SearchCommandStatus publishedStatus = await publishTask.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert + publishedStatus.ShouldBe(SearchCommandStatus.Completed); await _eventPublisher.Received(1).PublishSearchCompleted( - eventId, SearchCommandStatus.Failed, Arg.Any(), Arg.Any(), Arg.Any?>()); - - await _arrClient.DidNotReceive().GetCommandStatusAsync(Arg.Any(), Arg.Any()); + eventId, SearchCommandStatus.Completed, Arg.Any(), Arg.Any(), null); await WaitForTrackerCountAsync(0); } + [Fact] + public async Task Keeps_polling_when_an_individual_command_check_fails() + { + // Arrange + ArrInstance radarrInstance = TestDataContextFactory.AddRadarrInstance(_dataContext); + AddTracker(radarrInstance.Id, Guid.NewGuid()); + await _dataContext.SaveChangesAsync(); + await _eventsContext.SaveChangesAsync(); + + _arrClient.GetCommandsAsync(Arg.Any()) + .ThrowsAsync(new HttpRequestException("command list unavailable")); + _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new HttpRequestException("command status unavailable")); + + Task secondPoll = CaptureNthPoll(2); + CaptureNextPublishedStatus(); + + // Act + await _sut.StartAsync(_cts.Token); + await AdvanceUntilAsync(secondPoll); + + // Assert + await _eventPublisher.DidNotReceive().PublishSearchCompleted( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any?>()); + + SeekerCommandTracker tracker = _eventsContext.SeekerCommandTrackers.AsNoTracking().Single(); + tracker.Status.ShouldBe(SearchCommandStatus.Pending); + } + + private void StubCommandState(ArrCommandState state, long commandId = 1) + { + _arrClient.GetCommandsAsync(Arg.Any()) + .Returns([new ArrCommandStatus(commandId, state, null)]); + _arrClient.GetCommandStatusAsync(Arg.Any(), Arg.Any()) + .Returns(new ArrCommandStatus(commandId, state, null)); + } + private void AddTracker( Guid arrInstanceId, Guid eventId, @@ -603,7 +791,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable int seen = 0; _arrClient - .When(client => client.GetCommandStatusAsync(Arg.Any(), Arg.Any())) + .When(client => client.GetCommandsAsync(Arg.Any())) .Do(_ => { if (Interlocked.Increment(ref seen) >= count) @@ -622,7 +810,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable for (int attempt = 0; attempt < 100 && !signal.IsCompleted; attempt++) { await Task.Delay(10); - _timeProvider.Advance(TimeSpan.FromSeconds(15)); + _timeProvider.Advance(TimeSpan.FromSeconds(60)); } await signal.WaitAsync(TimeSpan.FromSeconds(5)); diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Arr/ArrClient.cs b/code/backend/Cleanuparr.Infrastructure/Features/Arr/ArrClient.cs index e8dfe8d9..08ca2de1 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Arr/ArrClient.cs +++ b/code/backend/Cleanuparr.Infrastructure/Features/Arr/ArrClient.cs @@ -270,6 +270,23 @@ public abstract class ArrClient : IArrClient return result ?? new ArrCommandStatus(commandId, ArrCommandState.Unknown, null); } + /// + public async Task> GetCommandsAsync(ArrInstance arrInstance) + { + UriBuilder uriBuilder = new(arrInstance.Url); + uriBuilder.Path = $"{uriBuilder.Path.TrimEnd('/')}/api/v3/command"; + + using HttpRequestMessage request = new(HttpMethod.Get, uriBuilder.Uri); + SetApiKey(request, arrInstance.ApiKey); + + using HttpResponseMessage response = await _httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead); + response.EnsureSuccessStatusCode(); + + List? result = await DeserializeStreamAsync>(response); + + return result ?? []; + } + protected abstract string GetSystemStatusUrlPath(); protected abstract string GetQueueUrlPath(); diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Arr/Interfaces/IArrClient.cs b/code/backend/Cleanuparr.Infrastructure/Features/Arr/Interfaces/IArrClient.cs index 80144ef6..806ebb2a 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Arr/Interfaces/IArrClient.cs +++ b/code/backend/Cleanuparr.Infrastructure/Features/Arr/Interfaces/IArrClient.cs @@ -36,6 +36,11 @@ public interface IArrClient /// Task GetCommandStatusAsync(ArrInstance arrInstance, long commandId); + /// + /// Gets every command the arr instance currently knows about + /// + Task> GetCommandsAsync(ArrInstance arrInstance); + bool IsRecordValid(QueueRecord record); /// diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/SeekerCommandMonitor.cs b/code/backend/Cleanuparr.Infrastructure/Features/Jobs/SeekerCommandMonitor.cs index a3e69070..e970a1b1 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/SeekerCommandMonitor.cs +++ b/code/backend/Cleanuparr.Infrastructure/Features/Jobs/SeekerCommandMonitor.cs @@ -20,7 +20,7 @@ namespace Cleanuparr.Infrastructure.Features.Jobs; /// public class SeekerCommandMonitor : BackgroundService { - private static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(15); + private static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(60); private static readonly TimeSpan IdleInterval = TimeSpan.FromSeconds(60); private static readonly TimeSpan CommandTimeout = TimeSpan.FromMinutes(10); private static readonly TimeSpan AbandonAfter = TimeSpan.FromMinutes(30); @@ -69,6 +69,7 @@ public class SeekerCommandMonitor : BackgroundService var dataContext = scope.ServiceProvider.GetRequiredService(); var eventsContext = scope.ServiceProvider.GetRequiredService(); var arrClientFactory = scope.ServiceProvider.GetRequiredService(); + var queueIterator = scope.ServiceProvider.GetRequiredService(); var eventPublisher = scope.ServiceProvider.GetRequiredService(); List trackers = await eventsContext.SeekerCommandTrackers @@ -84,6 +85,8 @@ public class SeekerCommandMonitor : BackgroundService DateTimeOffset now = _timeProvider.GetUtcNow(); bool didWork = false; + List toPoll = []; + foreach (SeekerCommandTracker tracker in trackers) { if (IsTerminal(tracker.Status)) @@ -97,41 +100,34 @@ public class SeekerCommandMonitor : BackgroundService continue; } - if (!instancesById.TryGetValue(tracker.ArrInstanceId, out ArrInstance? arrInstance)) + if (!instancesById.ContainsKey(tracker.ArrInstanceId)) { tracker.Status = SearchCommandStatus.Failed; continue; } + toPoll.Add(tracker); + } + + foreach (IGrouping group in toPoll.GroupBy(t => t.ArrInstanceId)) + { + ArrInstance arrInstance = instancesById[group.Key]; IArrClient arrClient = arrClientFactory.GetClient(arrInstance.ArrConfig.Type, arrInstance.Version); didWork = true; - try - { - ArrCommandStatus status = await arrClient.GetCommandStatusAsync(arrInstance, tracker.CommandId); - UpdateTrackerStatus(tracker, status); - } - catch (HttpRequestException ex) when (ex.StatusCode is HttpStatusCode.NotFound) - { - _logger.LogDebug( - "Command {CommandId} is no longer known to {Instance}, treating '{Title}' as completed", - tracker.CommandId, arrInstance.Name, tracker.ItemTitle); - tracker.Status = SearchCommandStatus.Completed; - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Failed to check command {CommandId} status on {Instance}", - tracker.CommandId, arrInstance.Name); - } + await PollInstanceCommandsAsync(arrClient, arrInstance, group.ToList()); } await eventsContext.SaveChangesAsync(stoppingToken); + Dictionary> queueSnapshots = await BuildQueueSnapshotsAsync( + trackers, instancesById, arrClientFactory, queueIterator); + foreach (SeekerCommandTracker tracker in trackers.Where(t => IsTerminal(t.Status))) { didWork = true; - if (await TryPublishOutcomeAsync(tracker, instancesById, arrClientFactory, eventPublisher)) + if (await TryPublishOutcomeAsync(tracker, instancesById, queueSnapshots, eventPublisher)) { eventsContext.SeekerCommandTrackers.Remove(tracker); continue; @@ -158,10 +154,137 @@ public class SeekerCommandMonitor : BackgroundService return didWork; } + private async Task PollInstanceCommandsAsync( + IArrClient arrClient, + ArrInstance arrInstance, + List trackers) + { + Dictionary? commands = await TryListCommandsAsync(arrClient, arrInstance); + + if (commands is null) + { + foreach (SeekerCommandTracker tracker in trackers) + { + await PollSingleCommandAsync(arrClient, arrInstance, tracker); + } + + return; + } + + foreach (SeekerCommandTracker tracker in trackers) + { + if (commands.TryGetValue(tracker.CommandId, out ArrCommandStatus? status)) + { + UpdateTrackerStatus(tracker, status); + continue; + } + + MarkForgottenCommandAsCompleted(tracker, arrInstance); + } + } + + private async Task?> TryListCommandsAsync(IArrClient arrClient, ArrInstance arrInstance) + { + try + { + List commands = await arrClient.GetCommandsAsync(arrInstance); + + return commands + .GroupBy(command => command.Id) + .ToDictionary(group => group.Key, group => group.First()); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Failed to list commands on {Instance}, falling back to checking each command individually", + arrInstance.Name); + return null; + } + } + + private async Task PollSingleCommandAsync(IArrClient arrClient, ArrInstance arrInstance, SeekerCommandTracker tracker) + { + try + { + ArrCommandStatus status = await arrClient.GetCommandStatusAsync(arrInstance, tracker.CommandId); + UpdateTrackerStatus(tracker, status); + } + catch (HttpRequestException ex) when (ex.StatusCode is HttpStatusCode.NotFound) + { + MarkForgottenCommandAsCompleted(tracker, arrInstance); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to check command {CommandId} status on {Instance}", + tracker.CommandId, arrInstance.Name); + } + } + + private void MarkForgottenCommandAsCompleted(SeekerCommandTracker tracker, ArrInstance arrInstance) + { + _logger.LogDebug( + "Command {CommandId} is no longer known to {Instance}, treating '{Title}' as completed", + tracker.CommandId, arrInstance.Name, tracker.ItemTitle); + tracker.Status = SearchCommandStatus.Completed; + } + + private async Task>> BuildQueueSnapshotsAsync( + List trackers, + Dictionary instancesById, + IArrClientFactory arrClientFactory, + IArrQueueIterator queueIterator) + { + Dictionary> snapshots = []; + + List instanceIds = trackers + .Where(tracker => tracker.Status is SearchCommandStatus.Completed) + .Select(tracker => tracker.ArrInstanceId) + .Distinct() + .Where(instancesById.ContainsKey) + .ToList(); + + foreach (Guid instanceId in instanceIds) + { + ArrInstance arrInstance = instancesById[instanceId]; + List records = []; + + try + { + IArrClient arrClient = arrClientFactory.GetClient(arrInstance.ArrConfig.Type, arrInstance.Version); + + await queueIterator.Iterate(arrClient, arrInstance, pageRecords => + { + records.AddRange(pageRecords); + return Task.CompletedTask; + }); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to inspect the download queue on {Instance}", arrInstance.Name); + } + + snapshots[instanceId] = records; + } + + return snapshots; + } + private async Task TryPublishOutcomeAsync( SeekerCommandTracker tracker, Dictionary instancesById, - IArrClientFactory arrClientFactory, + Dictionary> queueSnapshots, IEventPublisher eventPublisher) { try @@ -187,7 +310,8 @@ public class SeekerCommandMonitor : BackgroundService return true; } - List? grabbedItems = await InspectDownloadQueueAsync(tracker, arrInstance, arrClientFactory); + IReadOnlyList queue = queueSnapshots.GetValueOrDefault(tracker.ArrInstanceId, []); + List? grabbedItems = FindGrabbedItems(tracker, arrInstance, queue); await eventPublisher.PublishSearchCompleted(tracker.EventId, SearchCommandStatus.Completed, instanceType, instanceUrl, grabbedItems); _logger.LogDebug("Search command completed for event {EventId}", tracker.EventId); @@ -235,41 +359,29 @@ public class SeekerCommandMonitor : BackgroundService }; } - private async Task?> InspectDownloadQueueAsync( + private List? FindGrabbedItems( SeekerCommandTracker tracker, ArrInstance arrInstance, - IArrClientFactory arrClientFactory) + IReadOnlyList queue) { - try + List grabbedTitles = queue + .Where(r => arrInstance.ArrConfig.Type == InstanceType.Radarr + ? r.MovieId == tracker.ExternalItemId + : r.SeriesId == tracker.ExternalItemId + && (tracker.SeasonNumber == 0 || r.SeasonNumber == tracker.SeasonNumber)) + .Where(r => !string.IsNullOrEmpty(r.DownloadId)) + .GroupBy(r => r.DownloadId) + .Select(g => g.First()) + .Select(r => r.Title) + .ToList(); + + if (grabbedTitles.Count > 0) { - IArrClient arrClient = arrClientFactory.GetClient(arrInstance.ArrConfig.Type, arrInstance.Version); - - QueueListResponse queue = await arrClient.GetQueueItemsAsync(arrInstance, 1); - - var grabbedTitles = queue.Records - .Where(r => arrInstance.ArrConfig.Type == InstanceType.Radarr - ? r.MovieId == tracker.ExternalItemId - : r.SeriesId == tracker.ExternalItemId - && (tracker.SeasonNumber == 0 || r.SeasonNumber == tracker.SeasonNumber)) - .Where(r => !string.IsNullOrEmpty(r.DownloadId)) - .GroupBy(r => r.DownloadId) - .Select(g => g.First()) - .Select(r => r.Title) - .ToList(); - - if (grabbedTitles.Count > 0) - { - _logger.LogInformation("Search for '{Title}' on {Instance} grabbed {Count} items: {Items}", - tracker.ItemTitle, arrInstance.Name, grabbedTitles.Count, - string.Join(", ", grabbedTitles)); - } - - return grabbedTitles.Count > 0 ? grabbedTitles : null; - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Failed to inspect download queue after search completion"); - return null; + _logger.LogInformation("Search for '{Title}' on {Instance} grabbed {Count} items: {Items}", + tracker.ItemTitle, arrInstance.Name, grabbedTitles.Count, + string.Join(", ", grabbedTitles)); } + + return grabbedTitles.Count > 0 ? grabbedTitles : null; } }