Reduce the number of arr requests made while monitoring searches (#722)

This commit is contained in:
Flaminel committed 2026-08-16 01:03:54 +03:00
1 parent 5d18f203ad
commit 4b74aa23f3
5 files changed
+457 -78

No files matched your search

@@ -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<ArrCommandStatus> 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<ArrCommandStatus> 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<HttpRequestException>(() => _client.GetCommandsAsync(_arrInstance));
}
#endregion
#region ShouldRemoveFromQueue
[Fact]
@@ -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<ILogger<ArrQueueIterator>>()));
serviceProvider.GetService(typeof(IEventPublisher)).Returns(_eventPublisher);
var scope = Substitute.For<IServiceScope>();
@@ -92,8 +95,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
await _dataContext.SaveChangesAsync();
await _eventsContext.SaveChangesAsync();
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null));
StubCommandState(ArrCommandState.Completed);
// 3 episodes from same season pack share the same DownloadId
_arrClient.GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>())
@@ -148,8 +150,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
await _dataContext.SaveChangesAsync();
await _eventsContext.SaveChangesAsync();
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null));
StubCommandState(ArrCommandState.Completed);
// Queue has records with empty DownloadId and one valid record
_arrClient.GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>())
@@ -202,8 +203,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
await _dataContext.SaveChangesAsync();
await _eventsContext.SaveChangesAsync();
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null));
StubCommandState(ArrCommandState.Completed);
// Two different downloads for the same movie
_arrClient.GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>())
@@ -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<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(7, ArrCommandState.Started, null));
StubCommandState(ArrCommandState.Started, commandId: 7);
Task<SearchCommandStatus> publishTask = CaptureNextPublishedStatus();
@@ -284,8 +283,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
await _dataContext.SaveChangesAsync();
await _eventsContext.SaveChangesAsync();
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(1, ArrCommandState.Failed, null));
StubCommandState(ArrCommandState.Failed);
Task<SearchCommandStatus> publishTask = CaptureNextPublishedStatus();
@@ -312,8 +310,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
await _dataContext.SaveChangesAsync();
await _eventsContext.SaveChangesAsync();
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(1, state, null));
StubCommandState(state);
Task<SearchCommandStatus> publishTask = CaptureNextPublishedStatus();
@@ -336,6 +333,8 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
await _dataContext.SaveChangesAsync();
await _eventsContext.SaveChangesAsync();
_arrClient.GetCommandsAsync(Arg.Any<ArrInstance>())
.ThrowsAsync(new HttpRequestException("Not found", null, HttpStatusCode.NotFound));
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.ThrowsAsync(new HttpRequestException("Not found", null, HttpStatusCode.NotFound));
_arrClient.GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>())
@@ -362,8 +361,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
await _dataContext.SaveChangesAsync();
await _eventsContext.SaveChangesAsync();
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.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<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null));
StubCommandState(ArrCommandState.Completed);
_arrClient.GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>())
.Returns(new QueueListResponse());
@@ -414,8 +411,7 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
await _dataContext.SaveChangesAsync();
await _eventsContext.SaveChangesAsync();
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(1, ArrCommandState.Completed, null));
StubCommandState(ArrCommandState.Completed);
_arrClient.GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>())
.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<SearchCommandStatus> 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<InstanceType>(), Arg.Any<string>(), Arg.Any<List<string>?>());
await _arrClient.DidNotReceive().GetCommandsAsync(Arg.Any<ArrInstance>());
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<ArrInstance>())
.Returns(
[
new ArrCommandStatus(1, ArrCommandState.Completed, null),
new ArrCommandStatus(2, ArrCommandState.Completed, null),
new ArrCommandStatus(3, ArrCommandState.Completed, null),
]);
_arrClient.GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>())
.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<ArrInstance>());
await _arrClient.DidNotReceive().GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>());
await _arrClient.Received(1).GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>());
}
[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<ArrInstance>())
.Returns([new ArrCommandStatus(7, ArrCommandState.Started, null)]);
_arrClient.GetQueueItemsAsync(Arg.Any<ArrInstance>(), Arg.Any<int>())
.Returns(new QueueListResponse());
Task<SearchCommandStatus> 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<ArrInstance>())
.ThrowsAsync(new HttpRequestException("boom"));
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.Returns(new ArrCommandStatus(1, ArrCommandState.Failed, null));
Task<SearchCommandStatus> publishTask = CaptureNextPublishedStatus();
// Act
@@ -519,15 +606,116 @@ public class SeekerCommandMonitorTests : IAsyncDisposable
// Assert
publishedStatus.ShouldBe(SearchCommandStatus.Failed);
await _arrClient.Received(1).GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>());
}
[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<ArrInstance>(), 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<ArrInstance>(), 2)
.Returns(new QueueListResponse
{
TotalRecords = 2,
Records = [new QueueRecord { Id = 2, MovieId = 300, Title = "Wanted.Movie", DownloadId = "WANTED", Protocol = "torrent", Status = "downloading" }]
});
TaskCompletionSource<List<string>?> grabbedTcs = new();
_eventPublisher.PublishSearchCompleted(
Arg.Any<Guid>(), Arg.Any<SearchCommandStatus>(), Arg.Any<InstanceType>(), Arg.Any<string>(), Arg.Any<List<string>?>())
.Returns(Task.CompletedTask)
.AndDoes(ci => grabbedTcs.TrySetResult(ci.ArgAt<List<string>?>(4)));
// Act
await _sut.StartAsync(_cts.Token);
_timeProvider.Advance(TimeSpan.FromSeconds(11));
List<string>? 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<ArrInstance>(), Arg.Any<int>())
.ThrowsAsync(new HttpRequestException("queue unavailable"));
Task<SearchCommandStatus> 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<InstanceType>(), Arg.Any<string>(), Arg.Any<List<string>?>());
await _arrClient.DidNotReceive().GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>());
eventId, SearchCommandStatus.Completed, Arg.Any<InstanceType>(), Arg.Any<string>(), 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<ArrInstance>())
.ThrowsAsync(new HttpRequestException("command list unavailable"));
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.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<Guid>(), Arg.Any<SearchCommandStatus>(), Arg.Any<InstanceType>(), Arg.Any<string>(), Arg.Any<List<string>?>());
SeekerCommandTracker tracker = _eventsContext.SeekerCommandTrackers.AsNoTracking().Single();
tracker.Status.ShouldBe(SearchCommandStatus.Pending);
}
private void StubCommandState(ArrCommandState state, long commandId = 1)
{
_arrClient.GetCommandsAsync(Arg.Any<ArrInstance>())
.Returns([new ArrCommandStatus(commandId, state, null)]);
_arrClient.GetCommandStatusAsync(Arg.Any<ArrInstance>(), Arg.Any<long>())
.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<ArrInstance>(), Arg.Any<long>()))
.When(client => client.GetCommandsAsync(Arg.Any<ArrInstance>()))
.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));
@@ -270,6 +270,23 @@ public abstract class ArrClient : IArrClient
return result ?? new ArrCommandStatus(commandId, ArrCommandState.Unknown, null);
}
/// <inheritdoc/>
public async Task<List<ArrCommandStatus>> 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<ArrCommandStatus>? result = await DeserializeStreamAsync<List<ArrCommandStatus>>(response);
return result ?? [];
}
protected abstract string GetSystemStatusUrlPath();
protected abstract string GetQueueUrlPath();
@@ -36,6 +36,11 @@ public interface IArrClient
/// </summary>
Task<ArrCommandStatus> GetCommandStatusAsync(ArrInstance arrInstance, long commandId);
/// <summary>
/// Gets every command the arr instance currently knows about
/// </summary>
Task<List<ArrCommandStatus>> GetCommandsAsync(ArrInstance arrInstance);
bool IsRecordValid(QueueRecord record);
/// <summary>
@@ -20,7 +20,7 @@ namespace Cleanuparr.Infrastructure.Features.Jobs;
/// </summary>
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<DataContext>();
var eventsContext = scope.ServiceProvider.GetRequiredService<EventsContext>();
var arrClientFactory = scope.ServiceProvider.GetRequiredService<IArrClientFactory>();
var queueIterator = scope.ServiceProvider.GetRequiredService<IArrQueueIterator>();
var eventPublisher = scope.ServiceProvider.GetRequiredService<IEventPublisher>();
List<SeekerCommandTracker> trackers = await eventsContext.SeekerCommandTrackers
@@ -84,6 +85,8 @@ public class SeekerCommandMonitor : BackgroundService
DateTimeOffset now = _timeProvider.GetUtcNow();
bool didWork = false;
List<SeekerCommandTracker> 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<Guid, SeekerCommandTracker> 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<Guid, IReadOnlyList<QueueRecord>> 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<SeekerCommandTracker> trackers)
{
Dictionary<long, ArrCommandStatus>? 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<Dictionary<long, ArrCommandStatus>?> TryListCommandsAsync(IArrClient arrClient, ArrInstance arrInstance)
{
try
{
List<ArrCommandStatus> 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<Dictionary<Guid, IReadOnlyList<QueueRecord>>> BuildQueueSnapshotsAsync(
List<SeekerCommandTracker> trackers,
Dictionary<Guid, ArrInstance> instancesById,
IArrClientFactory arrClientFactory,
IArrQueueIterator queueIterator)
{
Dictionary<Guid, IReadOnlyList<QueueRecord>> snapshots = [];
List<Guid> 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<QueueRecord> 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<bool> TryPublishOutcomeAsync(
SeekerCommandTracker tracker,
Dictionary<Guid, ArrInstance> instancesById,
IArrClientFactory arrClientFactory,
Dictionary<Guid, IReadOnlyList<QueueRecord>> queueSnapshots,
IEventPublisher eventPublisher)
{
try
@@ -187,7 +310,8 @@ public class SeekerCommandMonitor : BackgroundService
return true;
}
List<string>? grabbedItems = await InspectDownloadQueueAsync(tracker, arrInstance, arrClientFactory);
IReadOnlyList<QueueRecord> queue = queueSnapshots.GetValueOrDefault(tracker.ArrInstanceId, []);
List<string>? 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<List<string>?> InspectDownloadQueueAsync(
private List<string>? FindGrabbedItems(
SeekerCommandTracker tracker,
ArrInstance arrInstance,
IArrClientFactory arrClientFactory)
IReadOnlyList<QueueRecord> queue)
{
try
List<string> 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;
}
}