The single serial DeliveryWorker is replaced by Infrastructure/Jobs: - Job rows are leased with one FindOneAndUpdate (oldest RunAt first, a two-minute lease) and a reaper returns expired leases every 30 s; - enqueueing wakes the workers, which otherwise poll every five seconds; - delivery runs eight at a time with at most two per host, so a slow or dead server holds two slots, not the queue; - a failure waits n^4 + 15 + jitter seconds (Mastodon's curve) for up to 16 attempts; a 4xx other than 408/429 is final, a 429 honours Retry-After; - RemoteInstance is a per-host circuit breaker: ten consecutive failures quarantine a host for an hour, doubling to a week, and its jobs wait without spending attempts; - a delivery is queued once per activity and inbox (unique DedupeKey), and finished jobs expire after seven days (TTL on FinishedAt). Migration _004 moves pending Delivery rows into jobs and marks them abandoned, so a rollback to the old worker cannot send them twice. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012CzABvBkbcFqoHdmi8b9WB
90 lines
3.1 KiB
C#
90 lines
3.1 KiB
C#
using Microsoft.AspNetCore.Builder;
|
|
using Microsoft.AspNetCore.Hosting;
|
|
using Microsoft.AspNetCore.Http;
|
|
using Microsoft.Extensions.Caching.Memory;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
|
|
using PrivaPub.Infrastructure.Http;
|
|
|
|
using System.Collections.Concurrent;
|
|
|
|
namespace PrivaPub.Tests.Support
|
|
{
|
|
public sealed class Peer : IAsyncDisposable
|
|
{
|
|
readonly WebApplication _app;
|
|
readonly ConcurrentDictionary<string, string> _documents = new();
|
|
readonly ConcurrentDictionary<string, (int Status, TimeSpan Delay)> _answers = new();
|
|
|
|
public int Port { get; }
|
|
public string A => $"http://127.0.0.1:{Port}";
|
|
public string B => $"http://localhost:{Port}";
|
|
public ConcurrentQueue<HttpRequestRecord> Requests { get; } = new();
|
|
|
|
Peer(WebApplication app, int port)
|
|
{
|
|
_app = app;
|
|
Port = port;
|
|
}
|
|
|
|
public static async Task<Peer> Start()
|
|
{
|
|
var builder = WebApplication.CreateSlimBuilder();
|
|
builder.WebHost.UseUrls("http://127.0.0.1:0");
|
|
var app = builder.Build();
|
|
Peer peer = default;
|
|
app.Run(async context =>
|
|
{
|
|
peer.Requests.Enqueue(new(context.Request.Method, context.Request.Path, context.Request.Headers["Signature"].ToString()));
|
|
var key = context.Request.Path.Value;
|
|
if (peer._answers.TryGetValue(key, out var answer))
|
|
{
|
|
if (answer.Delay > TimeSpan.Zero)
|
|
await Task.Delay(answer.Delay);
|
|
context.Response.StatusCode = answer.Status;
|
|
return;
|
|
}
|
|
if (!peer._documents.TryGetValue(key, out var document))
|
|
{
|
|
context.Response.StatusCode = StatusCodes.Status404NotFound;
|
|
return;
|
|
}
|
|
context.Response.ContentType = "application/activity+json";
|
|
await context.Response.WriteAsync(document.Replace("{A}", peer.A).Replace("{B}", peer.B));
|
|
});
|
|
await app.StartAsync();
|
|
peer = new Peer(app, new Uri(app.Urls.First()).Port);
|
|
return peer;
|
|
}
|
|
|
|
public void Serve(string path, string json) => _documents[path] = json;
|
|
|
|
public void Answer(string path, int status, TimeSpan delay = default) => _answers[path] = (status, delay);
|
|
|
|
public static FederationHttp Http(IMemoryCache cache = default)
|
|
{
|
|
var options = new FederationOptions { AllowPrivateNetworks = true, AllowPlainHttp = true };
|
|
var services = new ServiceCollection();
|
|
services.AddHttpClient(FederationHttp.ClientName)
|
|
.ConfigurePrimaryHttpMessageHandler(() => SafeHttpHandlerFactory.Create(options));
|
|
return new FederationHttp(services.BuildServiceProvider().GetRequiredService<IHttpClientFactory>(),
|
|
cache ?? new MemoryCache(new MemoryCacheOptions()), new StaticOptions<FederationOptions>(options),
|
|
NullLogger<FederationHttp>.Instance);
|
|
}
|
|
|
|
public async ValueTask DisposeAsync() => await _app.DisposeAsync();
|
|
}
|
|
|
|
public sealed record HttpRequestRecord(string Method, string Path, string Signature);
|
|
|
|
public sealed class StaticOptions<T> : IOptionsMonitor<T>
|
|
{
|
|
public StaticOptions(T value) => CurrentValue = value;
|
|
public T CurrentValue { get; }
|
|
public T Get(string name) => CurrentValue;
|
|
public IDisposable OnChange(Action<T, string> listener) => default;
|
|
}
|
|
}
|