76 lines
2.5 KiB
C#
76 lines
2.5 KiB
C#
 
|
using Microsoft.Extensions.Caching.Memory;
|
||
|
|
|
||
|
|
using MongoDB.Entities;
|
||
|
|
|
||
|
|
using PrivaPub.Models.Jobs;
|
||
|
|
|
||
|
|
namespace PrivaPub.Infrastructure.Jobs
|
||
|
|
{
|
||
|
|
public interface IHostCircuitBreaker
|
||
|
|
{
|
||
|
|
Task<DateTime?> UnavailableUntil(string host, CancellationToken token);
|
||
|
|
Task Succeeded(string host, CancellationToken token);
|
||
|
|
Task Failed(string host, string error, CancellationToken token);
|
||
|
|
}
|
||
|
|
|
||
|
|
public class HostCircuitBreaker : IHostCircuitBreaker
|
||
|
|
{
|
||
|
|
public const int Threshold = 10;
|
||
|
|
static readonly TimeSpan CacheLifetime = TimeSpan.FromSeconds(30);
|
||
|
|
|
||
|
|
readonly IMemoryCache _cache;
|
||
|
|
|
||
|
|
public HostCircuitBreaker(IMemoryCache cache)
|
||
|
|
{
|
||
|
|
_cache = cache;
|
||
|
|
}
|
||
|
|
|
||
|
|
public async Task<DateTime?> UnavailableUntil(string host, CancellationToken token)
|
||
|
|
{
|
||
|
|
var instance = await Instance(host, token);
|
||
|
|
return instance?.UnavailableUntil > DateTime.UtcNow ? instance.UnavailableUntil : default;
|
||
|
|
}
|
||
|
|
|
||
|
|
public async Task Succeeded(string host, CancellationToken token)
|
||
|
|
{
|
||
|
|
var instance = await Instance(host, token);
|
||
|
|
if (instance is not { ConsecutiveFailures: > 0 } && instance?.LastSuccessAt > DateTime.UtcNow.AddHours(-1))
|
||
|
|
return;
|
||
|
|
await DB.Default.Update<RemoteInstance>()
|
||
|
|
.Match(i => i.Host == host)
|
||
|
|
.Modify(i => i.ConsecutiveFailures, 0)
|
||
|
|
.Modify(i => i.UnavailableUntil, null)
|
||
|
|
.Modify(i => i.LastSuccessAt, DateTime.UtcNow)
|
||
|
|
.Option(o => o.IsUpsert = true)
|
||
|
|
.ExecuteAsync(token);
|
||
|
|
_cache.Remove(Key(host));
|
||
|
|
}
|
||
|
|
|
||
|
|
public async Task Failed(string host, string error, CancellationToken token)
|
||
|
|
{
|
||
|
|
var now = DateTime.UtcNow;
|
||
|
|
var instance = await DB.Default.UpdateAndGet<RemoteInstance>()
|
||
|
|
.Match(i => i.Host == host)
|
||
|
|
.Modify(b => b.Inc(i => i.ConsecutiveFailures, 1))
|
||
|
|
.Modify(i => i.LastFailureAt, now)
|
||
|
|
.Modify(i => i.LastError, error)
|
||
|
|
.Option(o => o.IsUpsert = true)
|
||
|
|
.ExecuteAsync(token);
|
||
|
|
var quarantine = Backoff.HostQuarantine(instance.ConsecutiveFailures, Threshold);
|
||
|
|
if (quarantine > TimeSpan.Zero)
|
||
|
|
await DB.Default.Update<RemoteInstance>().MatchID(instance.ID)
|
||
|
|
.Modify(i => i.UnavailableUntil, now + quarantine)
|
||
|
|
.ExecuteAsync(token);
|
||
|
|
_cache.Remove(Key(host));
|
||
|
|
}
|
||
|
|
|
||
|
|
async Task<RemoteInstance> Instance(string host, CancellationToken token) =>
|
||
|
|
await _cache.GetOrCreateAsync(Key(host), async entry =>
|
||
|
|
{
|
||
|
|
entry.AbsoluteExpirationRelativeToNow = CacheLifetime;
|
||
|
|
return await DB.Default.Find<RemoteInstance>().Match(i => i.Host == host).ExecuteFirstAsync(token);
|
||
|
|
});
|
||
|
|
|
||
|
|
static string Key(string host) => "remote-instance:" + host;
|
||
|
|
}
|
||
|
|
}
|