brock/torrentio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #110 from iPromKnight/hotfix-introduce-retry-policy-too

Browse Source 
retry polic and circuit breaker policy
This commit is contained in:
iPromKnight
2024-03-03 19:54:54 +00:00
committed by GitHub
parent a90462a112 c3a281c39f
commit 8298d4021f
3 changed files with 87 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
src/producer/Features/Crawlers/Torrentio/TorrentioCrawler.cs
View File

@@ -66,6 +66,11 @@ public partial class TorrentioCrawler(
var processedItemsCount = 0;
foreach (var item in items)
{
try
{
await state.RetryPolicy.ExecuteAsync(
async () =>
{
processedItemsCount++;
@@ -74,19 +79,20 @@ public partial class TorrentioCrawler(
if (waitTime > TimeSpan.Zero)
{
logger.LogInformation("Rate limit reached for {TorrentioInstance}", instance.Name);
logger.LogInformation("Waiting for {TorrentioInstance}: {WaitTime} seconds", instance.Name, waitTime / 1000.0);
logger.LogInformation(
"Waiting for {TorrentioInstance}: {WaitTime} seconds", instance.Name, waitTime / 1000.0);
await Task.Delay(waitTime);
}
if (processedItemsCount % 2 == 0)
{
var randomWait = Random.Shared.Next(1000, 5000);
logger.LogInformation("Waiting for {TorrentioInstance}: {WaitTime} seconds", instance.Name, randomWait / 1000.0);
logger.LogInformation(
"Waiting for {TorrentioInstance}: {WaitTime} seconds", instance.Name, randomWait / 1000.0);
await Task.Delay(randomWait);
}
try
{
await state.ResiliencyPolicy.ExecuteAsync(
async () =>
{
@@ -97,10 +103,13 @@ public partial class TorrentioCrawler(
newTorrents.AddRange(torrentInfo.Where(x => x != null).Select(x => x!));
}
});
});
}
catch (Exception error)
{
logger.LogError(error, "page processing error in TorrentioCrawler");
logger.LogInformation("Setting possibly rate limited for {TorrentioInstance}", instance.Name);
instance.SetPossiblyRateLimited(state);
}
}
@@ -115,7 +124,17 @@ public partial class TorrentioCrawler(
private void SetupResiliencyPolicyForInstance(TorrentioInstance instance, TorrentioScrapeInstance state)
{
var policy = Policy
var retryPolicy = Policy
.Handle<Exception>()
.WaitAndRetryAsync(
retryCount: 2, // initial attempt + 2 retries
sleepDurationProvider: retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
onRetry: (exception, timeSpan, retryCount, _) =>
{
logger.LogWarning($"Retry {retryCount} encountered an exception: {exception.Message}. Pausing for {timeSpan.Seconds} seconds instance {instance.Name}");
});
var circuitBreakerPolicy = Policy
.Handle<Exception>()
.CircuitBreakerAsync(
exceptionsAllowedBeforeBreaking: instance.RateLimit.ExceptionLimit,
@@ -127,26 +146,19 @@ public partial class TorrentioCrawler(
onReset: () => logger.LogInformation("Circuit closed for {TorrentioInstance}, calls will flow again", instance.Name),
onHalfOpen: () => logger.LogInformation("Circuit is half-open for {TorrentioInstance}, next call is a trial if it should close or break again", instance.Name));
var policyWrap = Policy.WrapAsync(policy, Policy.TimeoutAsync(TimeSpan.FromSeconds(30)));
var policyWrap = Policy.WrapAsync(circuitBreakerPolicy, Policy.TimeoutAsync(TimeSpan.FromSeconds(30)));
state.ResiliencyPolicy = policyWrap;
state.RetryPolicy = retryPolicy;
}
private async Task<List<Torrent?>?> ScrapeInstance(TorrentioInstance instance, string imdbId, HttpClient client)
{
logger.LogInformation("Searching Torrentio {TorrentioInstance}: {ImdbId}", instance.Name, imdbId);
try
{
var movieSlug = string.Format(MovieSlug, imdbId);
var urlSlug = string.Format(Url, movieSlug);
return await RunRequest(instance, urlSlug, imdbId, client);
}
catch (Exception error)
{
logger.LogError(error, "page processing error {TorrentioInstance}: {ImdbId}", instance.Name, imdbId);
throw;
}
}
private async Task<List<Torrent?>?> RunRequest(TorrentioInstance instance, string urlSlug, string imdbId, HttpClient client)
{
@@ -179,8 +191,6 @@ public partial class TorrentioCrawler(
}
private Torrent ParseTorrentDetails(string title, TorrentioInstance instance, string infoHash, string imdbId)
{
try
{
var torrent = new Torrent
{
@@ -215,10 +225,4 @@ public partial class TorrentioCrawler(
return torrent;
}
catch (Exception e)
{
logger.LogError(e, "Error parsing torrent details");
throw;
}
}
}

11
src/producer/Features/Crawlers/Torrentio/TorrentioInstanceExtensions.cs
View File

@@ -22,6 +22,17 @@ public static class TorrentioInstancesExtensions
return remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero;
}
public static void SetPossiblyRateLimited(this TorrentioInstance instance, TorrentioScrapeInstance state, int minutesToWait = 5)
{
// Set the start time to 15 minutes in the past so that the next check will result in a rate limit period of 15 minutes
var startedAt = DateTime.UtcNow.AddMinutes(-minutesToWait);
var requestCount = instance.RateLimit.RequestLimit;
// Update the scraper state for the instance
state.StartedAt = startedAt;
state.RequestCount = requestCount;
}
public static long TotalProcessedRequests(this TorrentioInstance instance, Dictionary<string, TorrentioScrapeInstance> scraperState) =>
!scraperState.TryGetValue(instance.Name, out var state) ? 0 : state.TotalProcessed;

1
src/producer/Features/Crawlers/Torrentio/TorrentioScrapeInstance.cs
View File

@@ -7,4 +7,5 @@ public class TorrentioScrapeInstance
public int TotalProcessed { get; set; }
public string? LastProcessedImdbId { get; set; }
public IAsyncPolicy? ResiliencyPolicy { get; set; }
public IAsyncPolicy? RetryPolicy { get; set; }
}