Implement new rate-limiting framework for torrent clients

This commit is contained in:
omgbeez 2026-01-17 12:22:10 -05:00 committed by Clifford Roche
parent 371a13e060
commit 0a7e85c3e2
20 changed files with 442 additions and 93 deletions

View file

@ -0,0 +1,4 @@
export interface RateLimitStatus {
nextDequeueTime: Date | null;
secondsRemaining: number;
}

View file

@ -13,6 +13,13 @@
<small>Last check: {{ diskSpaceStatus.lastCheckTime | date: 'short' }}</small> <small>Last check: {{ diskSpaceStatus.lastCheckTime | date: 'short' }}</small>
</div> </div>
} }
@if (rateLimitStatus?.nextDequeueTime) {
<div class="notification is-warning">
<strong>Debrid provider rate limit reached</strong>
<br />
New torrents will not be added until {{ rateLimitStatus.nextDequeueTime | date: 'medium' }}
</div>
}
<div class="table-container"> <div class="table-container">
<table class="table is-fullwidth is-hoverable"> <table class="table is-fullwidth is-hoverable">
<thead> <thead>

View file

@ -2,6 +2,7 @@ import { Component, OnInit } from '@angular/core';
import { Router } from '@angular/router'; import { Router } from '@angular/router';
import { Torrent } from '../models/torrent.model'; import { Torrent } from '../models/torrent.model';
import { DiskSpaceStatus } from '../models/disk-space-status.model'; import { DiskSpaceStatus } from '../models/disk-space-status.model';
import { RateLimitStatus } from '../models/rate-limit-status.model';
import { TorrentService } from '../torrent.service'; import { TorrentService } from '../torrent.service';
import { forkJoin, Observable } from 'rxjs'; import { forkJoin, Observable } from 'rxjs';
import { FormsModule } from '@angular/forms'; import { FormsModule } from '@angular/forms';
@ -21,8 +22,8 @@ export class TorrentTableComponent implements OnInit {
public torrents: Torrent[] = []; public torrents: Torrent[] = [];
public selectedTorrents: string[] = []; public selectedTorrents: string[] = [];
public error: string; public error: string;
public sortProperty = 'rdName'; public sortProperty = 'added';
public sortDirection: 'asc' | 'desc' = 'asc'; public sortDirection: 'asc' | 'desc' = 'desc';
public isDeleteModalActive: boolean; public isDeleteModalActive: boolean;
public deleteError: string; public deleteError: string;
@ -50,6 +51,7 @@ export class TorrentTableComponent implements OnInit {
public updateSettingsTorrentLifetime: number; public updateSettingsTorrentLifetime: number;
public diskSpaceStatus: DiskSpaceStatus | null = null; public diskSpaceStatus: DiskSpaceStatus | null = null;
public rateLimitStatus: RateLimitStatus | null = null;
constructor( constructor(
private router: Router, private router: Router,
@ -67,6 +69,16 @@ export class TorrentTableComponent implements OnInit {
this.diskSpaceStatus = status; this.diskSpaceStatus = status;
}); });
this.torrentService.getRateLimitStatus().subscribe({
next: (status) => {
this.rateLimitStatus = status;
},
});
this.torrentService.rateLimitStatus$.subscribe((status) => {
this.rateLimitStatus = status;
});
this.torrentService.update$.subscribe((result) => { this.torrentService.update$.subscribe((result) => {
this.torrents = result; this.torrents = result;
}); });

View file

@ -4,6 +4,7 @@ import * as signalR from '@microsoft/signalr';
import { Observable, Subject } from 'rxjs'; import { Observable, Subject } from 'rxjs';
import { Torrent, TorrentFileAvailability } from './models/torrent.model'; import { Torrent, TorrentFileAvailability } from './models/torrent.model';
import { DiskSpaceStatus } from './models/disk-space-status.model'; import { DiskSpaceStatus } from './models/disk-space-status.model';
import { RateLimitStatus } from './models/rate-limit-status.model';
import { APP_BASE_HREF } from '@angular/common'; import { APP_BASE_HREF } from '@angular/common';
@Injectable({ @Injectable({
@ -12,6 +13,7 @@ import { APP_BASE_HREF } from '@angular/common';
export class TorrentService { export class TorrentService {
public update$: Subject<Torrent[]> = new Subject(); public update$: Subject<Torrent[]> = new Subject();
public diskSpaceStatus$: Subject<DiskSpaceStatus> = new Subject(); public diskSpaceStatus$: Subject<DiskSpaceStatus> = new Subject();
public rateLimitStatus$: Subject<RateLimitStatus> = new Subject();
private connection: signalR.HubConnection; private connection: signalR.HubConnection;
@ -40,6 +42,10 @@ export class TorrentService {
this.diskSpaceStatus$.next(status); this.diskSpaceStatus$.next(status);
}); });
this.connection.on('rateLimitStatus', (status: any) => {
this.rateLimitStatus$.next(status);
});
this.connection.onreconnected(() => { this.connection.onreconnected(() => {
this.getDiskSpaceStatus().subscribe({ this.getDiskSpaceStatus().subscribe({
next: (status) => { next: (status) => {
@ -65,6 +71,10 @@ export class TorrentService {
return this.http.get<DiskSpaceStatus | null>(`${this.baseHref}Api/Torrents/DiskSpaceStatus`); return this.http.get<DiskSpaceStatus | null>(`${this.baseHref}Api/Torrents/DiskSpaceStatus`);
} }
public getRateLimitStatus(): Observable<RateLimitStatus | null> {
return this.http.get<RateLimitStatus | null>(`${this.baseHref}Api/Torrents/RateLimitStatus`);
}
public uploadMagnet(magnetLink: string, torrent: Torrent): Observable<void> { public uploadMagnet(magnetLink: string, torrent: Torrent): Observable<void> {
return this.http.post<void>(`${this.baseHref}Api/Torrents/UploadMagnet`, { return this.http.post<void>(`${this.baseHref}Api/Torrents/UploadMagnet`, {
magnetLink, magnetLink,

View file

@ -0,0 +1,6 @@
namespace RdtClient.Data.Models.Internal;
public class RateLimitException(String message, TimeSpan retryAfter) : Exception(message)
{
public TimeSpan RetryAfter { get; } = retryAfter;
}

View file

@ -0,0 +1,7 @@
namespace RdtClient.Data.Models.Internal;
public class RateLimitStatus
{
public DateTimeOffset? NextDequeueTime { get; set; }
public Double SecondsRemaining { get; set; }
}

View file

@ -0,0 +1,53 @@
using System.Net;
using RdtClient.Data.Models.Internal;
using RdtClient.Service.Helpers;
using Xunit;
namespace RdtClient.Service.Test.Helpers;
public class RateLimitHandlerTest
{
[Fact]
public async Task SendAsync_ThrowsRateLimitException_On429WithRetryAfter()
{
// Arrange
var handler = new RateLimitHandler
{
InnerHandler = new MockHttpMessageHandler(HttpStatusCode.TooManyRequests, 3600)
};
var client = new HttpClient(handler);
// Act & Assert
var ex = await Assert.ThrowsAsync<RateLimitException>(() => client.GetAsync("http://example.com"));
Assert.Equal(TimeSpan.FromSeconds(3600), ex.RetryAfter);
Assert.Equal("TorBox rate limit exceeded", ex.Message);
}
[Fact]
public async Task SendAsync_ThrowsRateLimitException_On429WithoutRetryAfter()
{
// Arrange
var handler = new RateLimitHandler
{
InnerHandler = new MockHttpMessageHandler(HttpStatusCode.TooManyRequests, null)
};
var client = new HttpClient(handler);
// Act & Assert
var ex = await Assert.ThrowsAsync<RateLimitException>(() => client.GetAsync("http://example.com"));
Assert.Equal(TimeSpan.FromMinutes(2), ex.RetryAfter);
}
private class MockHttpMessageHandler(HttpStatusCode statusCode, Int32? retryAfterSeconds) : HttpMessageHandler
{
protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
var response = new HttpResponseMessage(statusCode);
if (retryAfterSeconds.HasValue)
{
response.Headers.RetryAfter = new System.Net.Http.Headers.RetryConditionHeaderValue(TimeSpan.FromSeconds(retryAfterSeconds.Value));
}
return Task.FromResult(response);
}
}
}

View file

@ -20,6 +20,7 @@
<PackageReference Include="TestableIO.System.IO.Abstractions" Version="22.0.16" /> <PackageReference Include="TestableIO.System.IO.Abstractions" Version="22.0.16" />
<PackageReference Include="TestableIO.System.IO.Abstractions.TestingHelpers" Version="22.0.16" /> <PackageReference Include="TestableIO.System.IO.Abstractions.TestingHelpers" Version="22.0.16" />
<PackageReference Include="TestableIO.System.IO.Abstractions.Wrappers" Version="22.0.16" /> <PackageReference Include="TestableIO.System.IO.Abstractions.Wrappers" Version="22.0.16" />
<PackageReference Include="TorBox.NET" Version="1.6.2" />
<PackageReference Include="xunit" Version="2.9.3" /> <PackageReference Include="xunit" Version="2.9.3" />
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.5"> <PackageReference Include="xunit.runner.visualstudio" Version="3.1.5">
<PrivateAssets>all</PrivateAssets> <PrivateAssets>all</PrivateAssets>

View file

@ -2,6 +2,7 @@
using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using RdtClient.Data.Enums; using RdtClient.Data.Enums;
using RdtClient.Data.Models.Internal;
using RdtClient.Service.Services; using RdtClient.Service.Services;
namespace RdtClient.Service.BackgroundServices; namespace RdtClient.Service.BackgroundServices;
@ -19,6 +20,7 @@ public class ProviderUpdater(ILogger<ProviderUpdater> logger, IServiceProvider s
using var scope = serviceProvider.CreateScope(); using var scope = serviceProvider.CreateScope();
var torrentService = scope.ServiceProvider.GetRequiredService<Torrents>(); var torrentService = scope.ServiceProvider.GetRequiredService<Torrents>();
var torrentRunner = scope.ServiceProvider.GetRequiredService<TorrentRunner>();
logger.LogInformation("ProviderUpdater started."); logger.LogInformation("ProviderUpdater started.");
@ -56,6 +58,11 @@ public class ProviderUpdater(ILogger<ProviderUpdater> logger, IServiceProvider s
logger.LogDebug("Finished updating torrent info from debrid provider, next update in {updateTime} seconds", updateTime); logger.LogDebug("Finished updating torrent info from debrid provider, next update in {updateTime} seconds", updateTime);
} }
} }
catch (RateLimitException ex)
{
await torrentRunner.SetRateLimit(ex.RetryAfter, ex.Message);
_nextUpdate = DateTime.UtcNow.Add(ex.RetryAfter);
}
catch (Exception ex) catch (Exception ex)
{ {
logger.LogError(ex, "Unexpected error occurred in ProviderUpdater: {ex.Message}", ex.Message); logger.LogError(ex, "Unexpected error occurred in ProviderUpdater: {ex.Message}", ex.Message);

View file

@ -4,8 +4,11 @@ using System.Reflection;
using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Authorization;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Polly; using Polly;
using Polly.Extensions.Http; using Polly.Timeout;
using RateLimitHeaders.Polly;
using RdtClient.Data.Models.Internal;
using RdtClient.Service.BackgroundServices; using RdtClient.Service.BackgroundServices;
using RdtClient.Service.Helpers;
using RdtClient.Service.Middleware; using RdtClient.Service.Middleware;
using RdtClient.Service.Services; using RdtClient.Service.Services;
using RdtClient.Service.Services.TorrentClients; using RdtClient.Service.Services.TorrentClients;
@ -16,6 +19,7 @@ namespace RdtClient.Service;
public static class DiConfig public static class DiConfig
{ {
public const String RD_CLIENT = "RdClient"; public const String RD_CLIENT = "RdClient";
public const String TORBOX_CLIENT = "TorBoxClient";
public static readonly String UserAgent = $"rdt-client {Assembly.GetEntryAssembly()?.GetName().Version}"; public static readonly String UserAgent = $"rdt-client {Assembly.GetEntryAssembly()?.GetName().Version}";
public static void RegisterRdtServices(this IServiceCollection services) public static void RegisterRdtServices(this IServiceCollection services)
@ -58,11 +62,6 @@ public static class DiConfig
public static void RegisterHttpClients(this IServiceCollection services) public static void RegisterHttpClients(this IServiceCollection services)
{ {
var retryPolicy = HttpPolicyExtensions
.HandleTransientHttpError()
.OrResult(r => r.StatusCode == HttpStatusCode.TooManyRequests)
.WaitAndRetryAsync(retryCount: 5, sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)));
services.AddHttpClient(); services.AddHttpClient();
services.ConfigureHttpClientDefaults(builder => services.ConfigureHttpClientDefaults(builder =>
{ {
@ -72,7 +71,64 @@ public static class DiConfig
}); });
}); });
services.AddTransient<RateLimitHandler>();
services.AddHttpClient(RD_CLIENT) services.AddHttpClient(RD_CLIENT)
.AddPolicyHandler(retryPolicy); .AddHttpMessageHandler<RateLimitHandler>()
.AddResilienceHandler("rd_client_handler", ConfigureResiliencePipeline);
services.AddHttpClient(TORBOX_CLIENT)
.AddHttpMessageHandler<RateLimitHandler>()
.AddResilienceHandler("torbox_client_handler", ConfigureResiliencePipeline);
}
private static void ConfigureResiliencePipeline(ResiliencePipelineBuilder<HttpResponseMessage> builder)
{
builder.AddRateLimitHeaders(options =>
{
options.EnableProactiveThrottling = true;
});
builder.AddRetry(new()
{
ShouldHandle = args => args.Outcome switch
{
{ Exception: HttpRequestException } => PredicateResult.True(),
{ Result.StatusCode: HttpStatusCode.RequestTimeout } => PredicateResult.True(),
{ Result.StatusCode: HttpStatusCode.TooManyRequests } => PredicateResult.True(),
{ Result.StatusCode: >= HttpStatusCode.InternalServerError } => PredicateResult.True(),
_ => PredicateResult.False()
},
MaxRetryAttempts = 2,
BackoffType = DelayBackoffType.Exponential,
Delay = TimeSpan.FromSeconds(2),
UseJitter = true,
DelayGenerator = args =>
{
if (args.Outcome.Result is { StatusCode: HttpStatusCode.TooManyRequests } response)
{
var retryAfter = response.Headers.RetryAfter;
var delay = retryAfter?.Delta ?? (retryAfter?.Date.HasValue == true ? retryAfter.Date.Value - DateTimeOffset.UtcNow : TimeSpan.FromMinutes(2));
if (delay < TimeSpan.Zero)
{
delay = TimeSpan.FromMinutes(2);
}
if (delay >= TimeSpan.FromSeconds(Settings.Get.Provider.Timeout))
{
throw new RateLimitException("TorBox rate limit exceeded", delay);
}
return new ValueTask<TimeSpan?>(delay);
}
return new ValueTask<TimeSpan?>((TimeSpan?)null);
}
});
builder.AddTimeout(new TimeoutStrategyOptions
{
TimeoutGenerator = _ => new ValueTask<TimeSpan>(TimeSpan.FromSeconds(Settings.Get.Provider.Timeout))
});
} }
} }

View file

@ -0,0 +1,34 @@
using System.Net;
using RdtClient.Data.Models.Internal;
namespace RdtClient.Service.Helpers;
public class RateLimitHandler : DelegatingHandler
{
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
try
{
var response = await base.SendAsync(request, cancellationToken);
if (response.StatusCode == HttpStatusCode.TooManyRequests)
{
var retryAfter = response.Headers.RetryAfter;
var delay = retryAfter?.Delta ?? (retryAfter?.Date.HasValue == true ? retryAfter.Date.Value - DateTimeOffset.UtcNow : TimeSpan.FromMinutes(2));
if (delay < TimeSpan.Zero)
{
delay = TimeSpan.FromMinutes(2);
}
throw new RateLimitException("TorBox rate limit exceeded", delay);
}
return response;
}
catch (TaskCanceledException ex) when (!cancellationToken.IsCancellationRequested)
{
throw new RateLimitException("TorBox rate limit exceeded (timeout)", TimeSpan.FromMinutes(2));
}
}
}

View file

@ -18,6 +18,8 @@
<PackageReference Include="MonoTorrent" Version="3.0.2" /> <PackageReference Include="MonoTorrent" Version="3.0.2" />
<PackageReference Include="Polly" Version="8.6.4" /> <PackageReference Include="Polly" Version="8.6.4" />
<PackageReference Include="Premiumize.NET" Version="1.0.10" /> <PackageReference Include="Premiumize.NET" Version="1.0.10" />
<PackageReference Include="RateLimitHeaders" Version="1.0.0" />
<PackageReference Include="RateLimitHeaders.Polly" Version="1.0.0" />
<PackageReference Include="RD.NET" Version="2.1.11" /> <PackageReference Include="RD.NET" Version="2.1.11" />
<PackageReference Include="Serilog" Version="4.3.0" /> <PackageReference Include="Serilog" Version="4.3.0" />
<PackageReference Include="Serilog.Sinks.File" Version="7.0.0" /> <PackageReference Include="Serilog.Sinks.File" Version="7.0.0" />
@ -25,7 +27,7 @@
<PackageReference Include="Synology.Api.Client" Version="0.3.93" /> <PackageReference Include="Synology.Api.Client" Version="0.3.93" />
<PackageReference Include="TestableIO.System.IO.Abstractions" Version="22.0.16" /> <PackageReference Include="TestableIO.System.IO.Abstractions" Version="22.0.16" />
<PackageReference Include="TestableIO.System.IO.Abstractions.Wrappers" Version="22.0.16" /> <PackageReference Include="TestableIO.System.IO.Abstractions.Wrappers" Version="22.0.16" />
<PackageReference Include="TorBox.NET" Version="1.5.0" /> <PackageReference Include="TorBox.NET" Version="1.6.2" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>

View file

@ -1,4 +1,5 @@
using Microsoft.AspNetCore.SignalR; using Microsoft.AspNetCore.SignalR;
using RdtClient.Data.Models.Internal;
namespace RdtClient.Service.Services; namespace RdtClient.Service.Services;
@ -24,4 +25,9 @@ public class RemoteService(IHubContext<RdtHub> hub, Torrents torrents)
{ {
await hub.Clients.All.SendCoreAsync("diskSpaceStatus", [status]); await hub.Clients.All.SendCoreAsync("diskSpaceStatus", [status]);
} }
public async Task UpdateRateLimitStatus(RateLimitStatus status)
{
await hub.Clients.All.SendCoreAsync("rateLimitStatus", [status]);
}
} }

View file

@ -4,6 +4,7 @@ using Microsoft.Extensions.Logging;
using Newtonsoft.Json; using Newtonsoft.Json;
using RdtClient.Data.Enums; using RdtClient.Data.Enums;
using RdtClient.Data.Models.TorrentClient; using RdtClient.Data.Models.TorrentClient;
using RdtClient.Data.Models.Internal;
using RdtClient.Service.Helpers; using RdtClient.Service.Helpers;
using RdtClient.Data.Models.Data; using RdtClient.Data.Models.Data;
using File = AllDebridNET.File; using File = AllDebridNET.File;
@ -124,30 +125,46 @@ public class AllDebridTorrentClient(ILogger<AllDebridTorrentClient> logger, IAll
public async Task<String> AddMagnet(String magnetLink) public async Task<String> AddMagnet(String magnetLink)
{ {
var result = await allDebridNetClientFactory.GetClient().Magnet.UploadMagnetAsync(magnetLink); try
if (result?.Id == null)
{ {
throw new("Unable to add magnet link"); var result = await allDebridNetClientFactory.GetClient().Magnet.UploadMagnetAsync(magnetLink);
if (result?.Id == null)
{
throw new("Unable to add magnet link");
}
var resultId = result.Id.ToString() ?? throw new($"Invalid responseID {result.Id}");
return resultId;
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("rate limit exceeded", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
} }
var resultId = result.Id.ToString() ?? throw new($"Invalid responseID {result.Id}");
return resultId;
} }
public async Task<String> AddFile(Byte[] bytes) public async Task<String> AddFile(Byte[] bytes)
{ {
var result = await allDebridNetClientFactory.GetClient().Magnet.UploadFileAsync(bytes); try
if (result?.Id == null)
{ {
throw new("Unable to add torrent file"); var result = await allDebridNetClientFactory.GetClient().Magnet.UploadFileAsync(bytes);
if (result?.Id == null)
{
throw new("Unable to add torrent file");
}
var resultId = result.Id.ToString() ?? throw new($"Invalid responseID {result.Id}");
return resultId;
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("rate limit exceeded", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
} }
var resultId = result.Id.ToString() ?? throw new($"Invalid responseID {result.Id}");
return resultId;
} }
public Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash) public Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash)

View file

@ -4,6 +4,7 @@ using Newtonsoft.Json;
using DebridLinkFrNET; using DebridLinkFrNET;
using RdtClient.Data.Enums; using RdtClient.Data.Enums;
using RdtClient.Data.Models.TorrentClient; using RdtClient.Data.Models.TorrentClient;
using RdtClient.Data.Models.Internal;
using RdtClient.Service.Helpers; using RdtClient.Service.Helpers;
using RdtClient.Data.Models.Data; using RdtClient.Data.Models.Data;
using Download = RdtClient.Data.Models.Data.Download; using Download = RdtClient.Data.Models.Data.Download;
@ -120,16 +121,32 @@ public class DebridLinkClient(ILogger<DebridLinkClient> logger, IHttpClientFacto
public async Task<String> AddMagnet(String magnetLink) public async Task<String> AddMagnet(String magnetLink)
{ {
var result = await GetClient().Seedbox.AddTorrentAsync(magnetLink); try
{
var result = await GetClient().Seedbox.AddTorrentAsync(magnetLink);
return result.Id ?? ""; return result.Id ?? "";
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("rate limit exceeded", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
}
} }
public async Task<String> AddFile(Byte[] bytes) public async Task<String> AddFile(Byte[] bytes)
{ {
var result = await GetClient().Seedbox.AddTorrentByFileAsync(bytes); try
{
var result = await GetClient().Seedbox.AddTorrentByFileAsync(bytes);
return result.Id ?? ""; return result.Id ?? "";
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("rate limit exceeded", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
}
} }
public Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash) public Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash)

View file

@ -4,6 +4,7 @@ using Newtonsoft.Json;
using PremiumizeNET; using PremiumizeNET;
using RdtClient.Data.Enums; using RdtClient.Data.Enums;
using RdtClient.Data.Models.TorrentClient; using RdtClient.Data.Models.TorrentClient;
using RdtClient.Data.Models.Internal;
using RdtClient.Service.Helpers; using RdtClient.Service.Helpers;
using RdtClient.Data.Models.Data; using RdtClient.Data.Models.Data;
using Torrent = RdtClient.Data.Models.Data.Torrent; using Torrent = RdtClient.Data.Models.Data.Torrent;
@ -91,30 +92,46 @@ public class PremiumizeTorrentClient(ILogger<PremiumizeTorrentClient> logger, IH
public async Task<String> AddMagnet(String magnetLink) public async Task<String> AddMagnet(String magnetLink)
{ {
var result = await GetClient().Transfers.CreateAsync(magnetLink, ""); try
if (result?.Id == null)
{ {
throw new("Unable to add magnet link"); var result = await GetClient().Transfers.CreateAsync(magnetLink, "");
if (result?.Id == null)
{
throw new("Unable to add magnet link");
}
var resultId = result.Id ?? throw new($"Invalid responseID {result.Id}");
return resultId;
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("rate limit exceeded", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
} }
var resultId = result.Id ?? throw new($"Invalid responseID {result.Id}");
return resultId;
} }
public async Task<String> AddFile(Byte[] bytes) public async Task<String> AddFile(Byte[] bytes)
{ {
var result = await GetClient().Transfers.CreateAsync(bytes, ""); try
if (result?.Id == null)
{ {
throw new("Unable to add torrent file"); var result = await GetClient().Transfers.CreateAsync(bytes, "");
if (result?.Id == null)
{
throw new("Unable to add torrent file");
}
var resultId = result.Id ?? throw new($"Invalid responseID {result.Id}");
return resultId;
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("rate limit exceeded", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
} }
var resultId = result.Id ?? throw new($"Invalid responseID {result.Id}");
return resultId;
} }
public Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash) public Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash)

View file

@ -5,6 +5,7 @@ using RDNET;
using RdtClient.Data.Enums; using RdtClient.Data.Enums;
using RdtClient.Data.Models.Data; using RdtClient.Data.Models.Data;
using RdtClient.Data.Models.TorrentClient; using RdtClient.Data.Models.TorrentClient;
using RdtClient.Data.Models.Internal;
using RdtClient.Service.Helpers; using RdtClient.Service.Helpers;
using Download = RdtClient.Data.Models.Data.Download; using Download = RdtClient.Data.Models.Data.Download;
using Torrent = RDNET.Torrent; using Torrent = RDNET.Torrent;
@ -128,20 +129,36 @@ public class RealDebridTorrentClient(ILogger<RealDebridTorrentClient> logger, IH
public async Task<String> AddMagnet(String magnetLink) public async Task<String> AddMagnet(String magnetLink)
{ {
var timeoutCancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(Settings.Get.Provider.Timeout)); try
{
var timeoutCancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(Settings.Get.Provider.Timeout));
var result = await GetClient().Torrents.AddMagnetAsync(magnetLink, timeoutCancellationToken.Token); var result = await GetClient().Torrents.AddMagnetAsync(magnetLink, timeoutCancellationToken.Token);
return result.Id; return result.Id;
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("rate limit exceeded", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
}
} }
public async Task<String> AddFile(Byte[] bytes) public async Task<String> AddFile(Byte[] bytes)
{ {
var timeoutCancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(Settings.Get.Provider.Timeout)); try
{
var timeoutCancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(Settings.Get.Provider.Timeout));
var result = await GetClient().Torrents.AddFileAsync(bytes, timeoutCancellationToken.Token); var result = await GetClient().Torrents.AddFileAsync(bytes, timeoutCancellationToken.Token);
return result.Id; return result.Id;
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("rate limit exceeded", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
}
} }
public Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash) public Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash)

View file

@ -5,6 +5,7 @@ using TorBoxNET;
using RdtClient.Data.Enums; using RdtClient.Data.Enums;
using RdtClient.Data.Models.TorrentClient; using RdtClient.Data.Models.TorrentClient;
using RdtClient.Data.Models.Data; using RdtClient.Data.Models.Data;
using RdtClient.Data.Models.Internal;
using RdtClient.Service.Helpers; using RdtClient.Service.Helpers;
namespace RdtClient.Service.Services.TorrentClients; namespace RdtClient.Service.Services.TorrentClients;
@ -23,10 +24,10 @@ public class TorBoxTorrentClient(ILogger<TorBoxTorrentClient> logger, IHttpClien
throw new("TorBox API Key not set in the settings"); throw new("TorBox API Key not set in the settings");
} }
var httpClient = httpClientFactory.CreateClient(); var httpClient = httpClientFactory.CreateClient(DiConfig.TORBOX_CLIENT);
httpClient.Timeout = TimeSpan.FromSeconds(Settings.Get.Provider.Timeout); httpClient.Timeout = TimeSpan.FromSeconds(Settings.Get.Provider.Timeout);
var torBoxNetClient = new TorBoxNetClient(null, httpClient, 5); var torBoxNetClient = new TorBoxNetClient(null, httpClient, 1);
torBoxNetClient.UseApiAuthentication(apiKey); torBoxNetClient.UseApiAuthentication(apiKey);
// Get the server time to fix up the timezones on results // Get the server time to fix up the timezones on results
@ -120,35 +121,48 @@ public class TorBoxTorrentClient(ILogger<TorBoxTorrentClient> logger, IHttpClien
}; };
} }
private async Task<String> AddTorrentRetry(Func<Boolean, Task<String>> action)
{
try
{
return await action(false);
}
catch (RateLimitException)
{
throw;
}
catch (Exception ex) when (ex.InnerException is RateLimitException rateLimitException)
{
throw rateLimitException;
}
catch (TorBoxException ex) when (ex.Error.Equals("active_limit", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
}
catch (Exception ex) when (ex.Message.Contains("slow_down", StringComparison.OrdinalIgnoreCase))
{
throw new RateLimitException(ex.Message, TimeSpan.FromMinutes(2));
}
}
public async Task<String> AddMagnet(String magnetLink) public async Task<String> AddMagnet(String magnetLink)
{ {
var user = await GetClient().User.GetAsync(true); return await AddTorrentRetry(async asQueued =>
var result = await GetClient().Torrents.AddMagnetAsync(magnetLink, user.Data?.Settings?.SeedTorrents ?? 3, false);
if (result.Error == "ACTIVE_LIMIT")
{ {
var magnetLinkInfo = MonoTorrent.MagnetLink.Parse(magnetLink); var user = await GetClient().User.GetAsync(true);
return magnetLinkInfo.InfoHashes.V1!.ToHex().ToLowerInvariant(); var result = await GetClient().Torrents.AddMagnetAsync(magnetLink, user.Data?.Settings?.SeedTorrents ?? 3, as_queued: asQueued);
} return result.Data!.Hash!;
});
return result.Data!.Hash!;
} }
public async Task<String> AddFile(Byte[] bytes) public async Task<String> AddFile(Byte[] bytes)
{ {
var user = await GetClient().User.GetAsync(true); return await AddTorrentRetry(async asQueued =>
var result = await GetClient().Torrents.AddFileAsync(bytes, user.Data?.Settings?.SeedTorrents ?? 3);
if (result.Error == "ACTIVE_LIMIT")
{ {
using var stream = new MemoryStream(bytes); var user = await GetClient().User.GetAsync(true);
var result = await GetClient().Torrents.AddFileAsync(bytes, user.Data?.Settings?.SeedTorrents ?? 3, as_queued: asQueued);
var torrent = await MonoTorrent.Torrent.LoadAsync(stream); return result.Data!.Hash!;
return torrent.InfoHashes.V1!.ToHex().ToLowerInvariant(); });
}
return result.Data!.Hash!;
} }
public async Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash) public async Task<IList<TorrentClientAvailableFile>> GetAvailableFiles(String hash)

View file

@ -11,13 +11,15 @@ using System.Text.Json;
namespace RdtClient.Service.Services; namespace RdtClient.Service.Services;
public class TorrentRunner(ILogger<TorrentRunner> logger, Torrents torrents, Downloads downloads) public class TorrentRunner(ILogger<TorrentRunner> logger, Torrents torrents, Downloads downloads, RemoteService remoteService)
{ {
public static readonly ConcurrentDictionary<Guid, DownloadClient> ActiveDownloadClients = new(); public static readonly ConcurrentDictionary<Guid, DownloadClient> ActiveDownloadClients = new();
public static readonly ConcurrentDictionary<Guid, UnpackClient> ActiveUnpackClients = new(); public static readonly ConcurrentDictionary<Guid, UnpackClient> ActiveUnpackClients = new();
public static Boolean IsPausedForLowDiskSpace { get; set; } public static Boolean IsPausedForLowDiskSpace { get; set; }
public static DateTimeOffset NextDequeueTime { get; private set; } = DateTimeOffset.MinValue;
private readonly HttpClient _httpClient = new() private readonly HttpClient _httpClient = new()
{ {
Timeout = TimeSpan.FromSeconds(10) Timeout = TimeSpan.FromSeconds(10)
@ -323,27 +325,51 @@ public class TorrentRunner(ILogger<TorrentRunner> logger, Torrents torrents, Dow
if (torrentsToAddToProvider.Count != 0) if (torrentsToAddToProvider.Count != 0)
{ {
var downloadingTorrentsCount = allTorrents.Count(m => m.RdStatus is not (TorrentStatus.Queued or TorrentStatus.Finished or TorrentStatus.Error)); if (DateTimeOffset.Now < NextDequeueTime)
var maxParallelDownloads = Settings.Get.Provider.MaxParallelDownloads;
logger.LogDebug("Currently downloading {downloadingTorrentCount}/{maxParallelDownloads} torrents, {queuedCount} queued.",
downloadingTorrentsCount,
maxParallelDownloads,
torrentsToAddToProvider.Count);
var dequeueCount = maxParallelDownloads == 0 ? torrentsToAddToProvider.Count : maxParallelDownloads - downloadingTorrentsCount;
foreach (var torrent in torrentsToAddToProvider.Take(dequeueCount))
{ {
try logger.LogDebug($"Dequeuing torrents is paused until {NextDequeueTime}, {NextDequeueTime - DateTimeOffset.Now} remaining");
}
else
{
if (NextDequeueTime != DateTimeOffset.MinValue)
{ {
await torrents.DequeueFromDebridQueue(torrent); NextDequeueTime = DateTimeOffset.MinValue;
await remoteService.UpdateRateLimitStatus(new RateLimitStatus
{
NextDequeueTime = null,
SecondsRemaining = 0
});
} }
catch (Exception ex)
var downloadingTorrentsCount = allTorrents.Count(m => m.RdStatus is not (TorrentStatus.Queued or TorrentStatus.Finished or TorrentStatus.Error));
var maxParallelDownloads = Settings.Get.Provider.MaxParallelDownloads;
logger.LogDebug("Currently downloading {downloadingTorrentCount}/{maxParallelDownloads} torrents, {queuedCount} queued.",
downloadingTorrentsCount,
maxParallelDownloads,
torrentsToAddToProvider.Count);
var dequeueCount = maxParallelDownloads == 0 ? torrentsToAddToProvider.Count : maxParallelDownloads - downloadingTorrentsCount;
foreach (var torrent in torrentsToAddToProvider.Take(dequeueCount))
{ {
await torrents.UpdateComplete(torrent.TorrentId, $"Could not add to provider: {ex.Message}", DateTimeOffset.Now, true); try
logger.LogWarning(ex, "Could not dequeue torrent {torrentId}", torrent.TorrentId); {
await torrents.DequeueFromDebridQueue(torrent);
}
catch (RateLimitException ex)
{
await SetRateLimit(ex.RetryAfter, ex.Message);
break;
}
catch (Exception ex)
{
await torrents.UpdateComplete(torrent.TorrentId, $"Could not add to provider: {ex.Message}", DateTimeOffset.Now, true);
logger.LogWarning(ex, "Could not dequeue torrent {torrentId}", torrent.TorrentId);
}
} }
} }
} }
@ -698,6 +724,19 @@ public class TorrentRunner(ILogger<TorrentRunner> logger, Torrents torrents, Dow
} }
} }
public async Task SetRateLimit(TimeSpan retryAfter, String message)
{
NextDequeueTime = DateTimeOffset.Now.Add(retryAfter);
Log($"Rate-limit reached, pausing dequeuing for {retryAfter.TotalMinutes} minutes (until {NextDequeueTime}): {message}");
await remoteService.UpdateRateLimitStatus(new RateLimitStatus
{
NextDequeueTime = NextDequeueTime,
SecondsRemaining = retryAfter.TotalSeconds
});
}
private void Log(String message, Download? download, Torrent? torrent) private void Log(String message, Download? download, Torrent? torrent)
{ {
if (download != null) if (download != null)

View file

@ -2,6 +2,7 @@
using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc; using Microsoft.AspNetCore.Mvc;
using MonoTorrent; using MonoTorrent;
using RdtClient.Data.Models.Internal;
using RdtClient.Data.Models.TorrentClient; using RdtClient.Data.Models.TorrentClient;
using RdtClient.Service.Helpers; using RdtClient.Service.Helpers;
using RdtClient.Service.Services; using RdtClient.Service.Services;
@ -53,6 +54,28 @@ public class TorrentsController(ILogger<TorrentsController> logger, Torrents tor
return Ok(status); return Ok(status);
} }
[HttpGet]
[Route("RateLimitStatus")]
public ActionResult<RateLimitStatus> GetRateLimitStatus()
{
var nextDequeueTime = TorrentRunner.NextDequeueTime;
if (nextDequeueTime < DateTimeOffset.Now)
{
return Ok(new RateLimitStatus
{
NextDequeueTime = null,
SecondsRemaining = 0
});
}
return Ok(new RateLimitStatus
{
NextDequeueTime = nextDequeueTime,
SecondsRemaining = (nextDequeueTime - DateTimeOffset.Now).TotalSeconds
});
}
/// <summary> /// <summary>
/// Used for debugging only. Force a tick. /// Used for debugging only. Force a tick.
/// </summary> /// </summary>