Deliveries run on a Mongo job queue with leases, backoff and per-host limits

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

PrivaPub/Federation/Outbox/DeliveryService.cs

@@ -10,6 +10,10 @@ using System.Text.Json.Nodes;
 using PrivaPub.Federation.Actors;
 using PrivaPub.Federation.Signing;
 using PrivaPub.Infrastructure.Http;
+using PrivaPub.Infrastructure.Jobs;
+using PrivaPub.Models.Jobs;
+
+using System.Text.Json;
 
 namespace PrivaPub.Federation.Outbox
 {
@@ -20,31 +24,38 @@ namespace PrivaPub.Federation.Outbox
 		Task<IReadOnlyList<string>> FollowerInboxes(LocalActor actor, CancellationToken token);
 	}
 
+	public sealed record DeliveryPayload(string SignerId, LocalActorKind SignerKind, string Inbox, string Body);
+
 	public class DeliveryService : IDeliveryService
 	{
 		readonly DbEntities _dbEntities;
+		readonly IJobQueue _queue;
 
-		public DeliveryService(DbEntities dbEntities)
+		public DeliveryService(DbEntities dbEntities, IJobQueue queue)
 		{
 			_dbEntities = dbEntities;
+			_queue = queue;
 		}
 
 		public async Task Enqueue(LocalActor signer, IEnumerable<string> inboxes, JsonObject activity, CancellationToken token)
 		{
 			var body = activity.ToJsonString();
-			var deliveries = inboxes
+			var activityId = activity["id"] is JsonValue id && id.TryGetValue<string>(out var text) ? text : default;
+			var jobs = inboxes
 				.Where(i => !string.IsNullOrEmpty(i) && !i.StartsWith(signer.BaseAddress + "/", StringComparison.OrdinalIgnoreCase))
 				.Distinct(StringComparer.Ordinal)
-				.Select(inbox => new Delivery
+				.Select(inbox => Uri.TryCreate(inbox, UriKind.Absolute, out var uri) ? (inbox, uri) : default)
+				.Where(target => target.uri != default)
+				.Select(target => new Job
 				{
-					SignerId = signer.Id,
-					SignerKind = signer.Kind,
-					InboxURL = inbox,
-					Body = body
+					Kind = JobKind.Deliver,
+					Host = target.uri.Host.ToLowerInvariant(),
+					DedupeKey = activityId == default ? default : $"{activityId}|{target.inbox}",
+					Payload = JsonSerializer.Serialize(new DeliveryPayload(signer.Id, signer.Kind, target.inbox, body))
 				})
 				.ToList();
-			if (deliveries.Count > 0)
-				await DB.Default.SaveAsync(deliveries, token);
+			if (jobs.Count > 0)
+				await _queue.EnqueueMany(jobs, token);
 		}
 
 		public async Task EnqueueToFollowers(LocalActor signer, JsonObject activity, CancellationToken token, IEnumerable<string> extraInboxes = default)
@@ -64,118 +75,81 @@ namespace PrivaPub.Federation.Outbox
 		}
 	}
 
-	public class DeliveryWorker : BackgroundService
+	public class DeliveryJobHandler : IJobHandler
 	{
-		const int MaxAttempts = 8;
-		static readonly TimeSpan Poll = TimeSpan.FromSeconds(3);
-
-		readonly IServiceProvider _services;
+		readonly ILocalActorService _actors;
 		readonly IFederationHttp _http;
-		readonly ILogger<DeliveryWorker> _logger;
+		readonly IHostCircuitBreaker _breaker;
+		readonly ILogger<DeliveryJobHandler> _logger;
 
-		public DeliveryWorker(IServiceProvider services, IFederationHttp http, ILogger<DeliveryWorker> logger)
+		public DeliveryJobHandler(ILocalActorService actors, IFederationHttp http, IHostCircuitBreaker breaker, ILogger<DeliveryJobHandler> logger)
 		{
-			_services = services;
+			_actors = actors;
 			_http = http;
+			_breaker = breaker;
 			_logger = logger;
 		}
 
-		protected override async Task ExecuteAsync(CancellationToken stoppingToken)
-		{
-			while (!stoppingToken.IsCancellationRequested)
-			{
-				try
-				{
-					await DeliverDue(stoppingToken);
-				}
-				catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
-				{
-					return;
-				}
-				catch (Exception ex)
-				{
-					_logger.LogError(ex, "{Worker} pass failed", nameof(DeliveryWorker));
-				}
-				await Task.Delay(Poll, stoppingToken);
-			}
-		}
+		public JobKind Kind => JobKind.Deliver;
+		public int Concurrency => 8;
+		public int MaxAttempts => 16;
+		public int PerHostLimit => 2;
 
-		async Task DeliverDue(CancellationToken token)
+		public async Task<JobOutcome> Handle(Job job, CancellationToken token)
 		{
-			using var scope = _services.CreateScope();
-			var dbEntities = scope.ServiceProvider.GetRequiredService<DbEntities>();
-			var actors = scope.ServiceProvider.GetRequiredService<ILocalActorService>();
-			var now = DateTime.UtcNow;
-			var due = await dbEntities.Deliveries
-				.Match(d => !d.DeliveredAt.HasValue && !d.AbandonedAt.HasValue && d.NextAttemptAt <= now)
-				.Sort(d => d.NextAttemptAt, Order.Ascending)
-				.Limit(20)
-				.ExecuteAsync(token);
+			var payload = JsonSerializer.Deserialize<DeliveryPayload>(job.Payload);
+			if (payload == default || !Uri.TryCreate(payload.Inbox, UriKind.Absolute, out var inbox) || !_http.IsAllowed(inbox))
+				return JobOutcome.Dead("not a deliverable inbox");
 
-			foreach (var delivery in due)
-			{
-				var signer = await actors.FindById(delivery.SignerKind, delivery.SignerId, token);
-				if (signer == default)
-				{
-					delivery.AbandonedAt = DateTime.UtcNow;
-					delivery.LastError = "the signing actor no longer exists";
-					await DB.Default.SaveAsync(delivery, token);
-					continue;
-				}
-				await Deliver(delivery, signer, token);
-				await DB.Default.SaveAsync(delivery, token);
-			}
-		}
+			var unavailableUntil = await _breaker.UnavailableUntil(job.Host, token);
+			if (unavailableUntil.HasValue)
+				return JobOutcome.Defer(unavailableUntil.Value, "the host is unavailable");
+
+			var signer = await _actors.FindById(payload.SignerKind, payload.SignerId, token);
+			if (signer == default)
+				return JobOutcome.Dead("the signing actor no longer exists");
+
+			var body = Encoding.UTF8.GetBytes(payload.Body);
+			using var request = new HttpRequestMessage(HttpMethod.Post, inbox) { Content = new ByteArrayContent(body) };
+			request.Content.Headers.ContentType = MediaTypeHeaderValue.Parse(RemoteActorService.ActivityJson);
+			HttpSignatures.Sign(request, signer, body);
 
-		async Task Deliver(Delivery delivery, LocalActor signer, CancellationToken token)
-		{
-			delivery.Attempts++;
 			try
 			{
-				if (!Uri.TryCreate(delivery.InboxURL, UriKind.Absolute, out var inbox) || !_http.IsAllowed(inbox))
-				{
-					delivery.AbandonedAt = DateTime.UtcNow;
-					delivery.LastError = "not a deliverable inbox";
-					return;
-				}
-
-				var body = Encoding.UTF8.GetBytes(delivery.Body);
-				using var request = new HttpRequestMessage(HttpMethod.Post, inbox)
-				{
-					Content = new ByteArrayContent(body)
-				};
-				request.Content.Headers.ContentType = MediaTypeHeaderValue.Parse(RemoteActorService.ActivityJson);
-				HttpSignatures.Sign(request, signer, body);
-
 				using var response = await _http.Send(request, token);
+				var status = (int)response.StatusCode;
 				if (response.IsSuccessStatusCode)
 				{
-					delivery.DeliveredAt = DateTime.UtcNow;
-					delivery.LastError = default;
-					return;
+					await _breaker.Succeeded(job.Host, token);
+					return JobOutcome.Done;
 				}
-
-				delivery.LastError = $"{(int)response.StatusCode} {response.ReasonPhrase}";
-				if (response.StatusCode is HttpStatusCode.Gone or HttpStatusCode.NotFound or HttpStatusCode.BadRequest or HttpStatusCode.Forbidden)
+				if (response.StatusCode == HttpStatusCode.TooManyRequests)
 				{
-					delivery.AbandonedAt = DateTime.UtcNow;
-					_logger.LogWarning("Delivery {Id} to {Inbox} abandoned: {Status}", delivery.ID, delivery.InboxURL, delivery.LastError);
-					return;
+					var retryAfter = response.Headers.RetryAfter?.Delta ?? (response.Headers.RetryAfter?.Date - DateTimeOffset.UtcNow);
+					return retryAfter > TimeSpan.Zero
+						? JobOutcome.Defer(DateTime.UtcNow + Min(retryAfter.Value, TimeSpan.FromHours(6)), "429")
+						: JobOutcome.Retry("429");
 				}
+				if (status is >= 400 and < 500 && status != 408)
+				{
+					await _breaker.Succeeded(job.Host, token);
+					_logger.LogInformation("Delivery to {Inbox} refused with {Status}", payload.Inbox, status);
+					return JobOutcome.Dead($"{status} {response.ReasonPhrase}");
+				}
+				await _breaker.Failed(job.Host, $"{status}", token);
+				return JobOutcome.Retry($"{status} {response.ReasonPhrase}");
 			}
-			catch (Exception ex) when (ex is HttpRequestException or BlockedDestinationException or TaskCanceledException && !token.IsCancellationRequested)
+			catch (BlockedDestinationException ex)
 			{
-				delivery.LastError = ex.Message;
+				return JobOutcome.Dead(ex.Message);
 			}
-
-			if (delivery.Attempts >= MaxAttempts)
+			catch (Exception ex) when (ex is HttpRequestException or TaskCanceledException && !token.IsCancellationRequested)
 			{
-				delivery.AbandonedAt = DateTime.UtcNow;
-				_logger.LogWarning("Delivery {Id} to {Inbox} abandoned after {Attempts} attempts: {Error}",
-					delivery.ID, delivery.InboxURL, delivery.Attempts, delivery.LastError);
-				return;
+				await _breaker.Failed(job.Host, ex.GetType().Name, token);
+				return JobOutcome.Retry(ex.Message);
 			}
-			delivery.NextAttemptAt = DateTime.UtcNow.AddMinutes(Math.Pow(2, delivery.Attempts));
 		}
+
+		static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b;
 	}
 }

PrivaPub/Infrastructure/Data/Indexes.cs

@@ -4,6 +4,7 @@ using MongoDB.Entities;
 
 using PrivaPub.Models.Federation;
 using PrivaPub.Models.Group;
+using PrivaPub.Models.Jobs;
 using PrivaPub.Models.Post;
 using PrivaPub.Models.User;
 
@@ -48,6 +49,15 @@ namespace PrivaPub.Infrastructure.Data
 			await Plain<DmGroup>(token, g => g.ParticipantsKey);
 
 			await Plain<Delivery>(token, d => d.DeliveredAt, d => d.AbandonedAt, d => d.NextAttemptAt);
+
+			await Plain<Job>(token, j => j.Kind, j => j.State, j => j.RunAt);
+			await Plain<Job>(token, j => j.State, j => j.LeasedUntil);
+			await Unique<Job>(j => j.DedupeKey, Builders<Job>.Filter.Type(j => j.DedupeKey, BsonType.String), token);
+			await DB.Default.Index<Job>()
+				.Key(j => j.FinishedAt, KeyType.Ascending)
+				.Option(o => o.ExpireAfter = TimeSpan.FromDays(7))
+				.CreateAsync(token);
+			await Unique<RemoteInstance>(i => i.Host, Builders<RemoteInstance>.Filter.Type(i => i.Host, BsonType.String), token);
 		}
 
 		static async Task Unique<T>(System.Linq.Expressions.Expression<Func<T, object>> key, FilterDefinition<T> partial,

PrivaPub/Infrastructure/Data/Migrations/_004_pending_deliveries_become_jobs.cs

@@ -0,0 +1,43 @@
+using MongoDB.Entities;
+
+using PrivaPub.Federation.Outbox;
+using PrivaPub.Models.Federation;
+using PrivaPub.Models.Jobs;
+
+using System.Text.Json;
+using System.Text.Json.Nodes;
+
+namespace PrivaPub.Infrastructure.Data.Migrations
+{
+	public class _004_pending_deliveries_become_jobs : IMigration
+	{
+		public async Task UpgradeAsync()
+		{
+			var pending = await DB.Default.Find<Delivery>()
+				.Match(d => !d.DeliveredAt.HasValue && !d.AbandonedAt.HasValue)
+				.ExecuteAsync();
+			foreach (var delivery in pending)
+			{
+				if (Uri.TryCreate(delivery.InboxURL, UriKind.Absolute, out var inbox))
+				{
+					var activityId = JsonNode.Parse(delivery.Body)?["id"]?.GetValue<string>();
+					var job = new Job
+					{
+						Kind = JobKind.Deliver,
+						Host = inbox.Host.ToLowerInvariant(),
+						DedupeKey = activityId == default ? default : $"{activityId}|{delivery.InboxURL}",
+						Attempts = delivery.Attempts,
+						RunAt = delivery.NextAttemptAt,
+						Payload = JsonSerializer.Serialize(new DeliveryPayload(delivery.SignerId, delivery.SignerKind, delivery.InboxURL, delivery.Body))
+					};
+					if (!await DB.Default.Find<Job>().Match(j => j.DedupeKey == job.DedupeKey && job.DedupeKey != null).ExecuteAnyAsync())
+						await DB.Default.SaveAsync(job);
+				}
+				await DB.Default.Update<Delivery>().MatchID(delivery.ID)
+					.Modify(d => d.AbandonedAt, DateTime.UtcNow)
+					.Modify(d => d.LastError, "moved to the job queue")
+					.ExecuteAsync();
+			}
+		}
+	}
+}

PrivaPub/Infrastructure/Jobs/Backoff.cs

@@ -0,0 +1,13 @@
+namespace PrivaPub.Infrastructure.Jobs
+{
+	public static class Backoff
+	{
+		public static TimeSpan After(int attempt) =>
+			TimeSpan.FromSeconds(Math.Pow(attempt, 4) + 15 + Random.Shared.Next(10) * (attempt + 1));
+
+		public static TimeSpan HostQuarantine(int consecutiveFailures, int threshold) =>
+			consecutiveFailures < threshold
+				? TimeSpan.Zero
+				: TimeSpan.FromHours(Math.Min(Math.Pow(2, consecutiveFailures - threshold), 24 * 7));
+	}
+}

PrivaPub/Infrastructure/Jobs/HostCircuitBreaker.cs

@@ -0,0 +1,76 @@
+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;
+	}
+}

PrivaPub/Infrastructure/Jobs/JobQueue.cs

@@ -0,0 +1,142 @@
+using MongoDB.Driver;
+using MongoDB.Entities;
+
+using PrivaPub.Models.Jobs;
+
+using System.Collections.Concurrent;
+
+namespace PrivaPub.Infrastructure.Jobs
+{
+	public sealed record JobOutcome(JobResult Result, string Error = default, DateTime? RetryAt = default)
+	{
+		public static readonly JobOutcome Done = new(JobResult.Done);
+		public static JobOutcome Retry(string error) => new(JobResult.Retry, error);
+		public static JobOutcome Dead(string error) => new(JobResult.Dead, error);
+		public static JobOutcome Defer(DateTime until, string error) => new(JobResult.Defer, error, until);
+	}
+
+	public enum JobResult
+	{
+		Done,
+		Retry,
+		Dead,
+		Defer
+	}
+
+	public interface IJobQueue
+	{
+		Task<bool> Enqueue(JobKind kind, string payload, string host, string dedupeKey, CancellationToken token);
+		Task<int> EnqueueMany(IEnumerable<Job> jobs, CancellationToken token);
+		Task<Job> Lease(JobKind kind, IReadOnlyCollection<string> busyHosts, CancellationToken token);
+		Task Finish(Job job, JobOutcome outcome, int maxAttempts, CancellationToken token);
+		Task<long> Reap(CancellationToken token);
+		Task WaitForWork(JobKind kind, TimeSpan poll, CancellationToken token);
+	}
+
+	public class JobQueue : IJobQueue
+	{
+		public static readonly TimeSpan LeaseTime = TimeSpan.FromMinutes(2);
+
+		readonly string _owner = $"{Environment.MachineName}:{Environment.ProcessId}";
+		readonly ConcurrentDictionary<JobKind, SemaphoreSlim> _signals = new();
+
+		public async Task<bool> Enqueue(JobKind kind, string payload, string host, string dedupeKey, CancellationToken token) =>
+			await EnqueueMany(new[] { new Job { Kind = kind, Payload = payload, Host = host, DedupeKey = dedupeKey } }, token) == 1;
+
+		public async Task<int> EnqueueMany(IEnumerable<Job> jobs, CancellationToken token)
+		{
+			var inserted = 0;
+			foreach (var job in jobs)
+			{
+				try
+				{
+					await DB.Default.SaveAsync(job, token);
+					inserted++;
+					Signal(job.Kind);
+				}
+				catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
+				{
+				}
+			}
+			return inserted;
+		}
+
+		public async Task<Job> Lease(JobKind kind, IReadOnlyCollection<string> busyHosts, CancellationToken token)
+		{
+			var now = DateTime.UtcNow;
+			var busy = busyHosts?.ToList() ?? new List<string>();
+			return await DB.Default.UpdateAndGet<Job>()
+				.Match(j => j.Kind == kind && j.State == JobState.Pending && j.RunAt <= now && !busy.Contains(j.Host))
+				.Modify(j => j.State, JobState.Running)
+				.Modify(j => j.LeasedUntil, now + LeaseTime)
+				.Modify(j => j.LeaseOwner, _owner)
+				.Modify(b => b.Inc(j => j.Attempts, 1))
+				.Option(o => o.Sort = Builders<Job>.Sort.Ascending(j => j.RunAt))
+				.ExecuteAsync(token);
+		}
+
+		public async Task Finish(Job job, JobOutcome outcome, int maxAttempts, CancellationToken token)
+		{
+			var now = DateTime.UtcNow;
+			var update = DB.Default.Update<Job>().MatchID(job.ID)
+				.Modify(j => j.LeasedUntil, null)
+				.Modify(j => j.LeaseOwner, null)
+				.Modify(j => j.LastError, outcome.Error);
+			switch (outcome.Result)
+			{
+				case JobResult.Done:
+					update.Modify(j => j.State, JobState.Done).Modify(j => j.FinishedAt, now);
+					break;
+				case JobResult.Retry when job.Attempts < maxAttempts:
+					update.Modify(j => j.State, JobState.Pending).Modify(j => j.RunAt, now + Backoff.After(job.Attempts));
+					break;
+				case JobResult.Defer:
+					update.Modify(j => j.State, JobState.Pending)
+						.Modify(j => j.RunAt, outcome.RetryAt ?? now + Backoff.After(job.Attempts))
+						.Modify(b => b.Inc(j => j.Attempts, -1));
+					break;
+				default:
+					update.Modify(j => j.State, JobState.Dead).Modify(j => j.FinishedAt, now);
+					break;
+			}
+			await update.ExecuteAsync(token);
+		}
+
+		public async Task<long> Reap(CancellationToken token)
+		{
+			var now = DateTime.UtcNow;
+			var result = await DB.Default.Update<Job>()
+				.Match(j => j.State == JobState.Running && j.LeasedUntil < now)
+				.Modify(j => j.State, JobState.Pending)
+				.Modify(j => j.RunAt, now)
+				.Modify(j => j.LeasedUntil, null)
+				.Modify(j => j.LeaseOwner, null)
+				.ExecuteAsync(token);
+			return result.ModifiedCount;
+		}
+
+		public async Task WaitForWork(JobKind kind, TimeSpan poll, CancellationToken token)
+		{
+			try
+			{
+				await SignalFor(kind).WaitAsync(poll, token);
+			}
+			catch (OperationCanceledException) when (token.IsCancellationRequested)
+			{
+			}
+		}
+
+		void Signal(JobKind kind)
+		{
+			try
+			{
+				SignalFor(kind).Release();
+			}
+			catch (SemaphoreFullException)
+			{
+			}
+		}
+
+		SemaphoreSlim SignalFor(JobKind kind) => _signals.GetOrAdd(kind, _ => new SemaphoreSlim(0, 64));
+	}
+}

PrivaPub/Infrastructure/Jobs/JobWorker.cs

@@ -0,0 +1,137 @@
+using PrivaPub.Models.Jobs;
+
+using System.Collections.Concurrent;
+
+namespace PrivaPub.Infrastructure.Jobs
+{
+	public interface IJobHandler
+	{
+		JobKind Kind { get; }
+		int Concurrency { get; }
+		int MaxAttempts { get; }
+		int PerHostLimit { get; }
+		Task<JobOutcome> Handle(Job job, CancellationToken token);
+	}
+
+	public class JobWorker : BackgroundService
+	{
+		static readonly TimeSpan Poll = TimeSpan.FromSeconds(5);
+		static readonly TimeSpan ReapInterval = TimeSpan.FromSeconds(30);
+
+		readonly IJobQueue _queue;
+		readonly IEnumerable<IJobHandler> _handlers;
+		readonly ILogger<JobWorker> _logger;
+
+		public JobWorker(IJobQueue queue, IEnumerable<IJobHandler> handlers, ILogger<JobWorker> logger)
+		{
+			_queue = queue;
+			_handlers = handlers;
+			_logger = logger;
+		}
+
+		protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+		{
+			var loops = _handlers.SelectMany(handler =>
+			{
+				var inFlight = new ConcurrentDictionary<string, int>(StringComparer.OrdinalIgnoreCase);
+				return Enumerable.Range(0, handler.Concurrency).Select(_ => Run(handler, inFlight, stoppingToken));
+			}).Append(Reap(stoppingToken));
+			await Task.WhenAll(loops);
+		}
+
+		async Task Run(IJobHandler handler, ConcurrentDictionary<string, int> inFlight, CancellationToken stoppingToken)
+		{
+			while (!stoppingToken.IsCancellationRequested)
+			{
+				Job job;
+				try
+				{
+					var busy = inFlight.Where(h => h.Value >= handler.PerHostLimit).Select(h => h.Key).ToList();
+					job = await _queue.Lease(handler.Kind, busy, stoppingToken);
+				}
+				catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+				{
+					return;
+				}
+				catch (Exception ex)
+				{
+					_logger.LogError(ex, "{Worker} could not lease a {Kind} job", nameof(JobWorker), handler.Kind);
+					await Delay(Poll, stoppingToken);
+					continue;
+				}
+
+				if (job == default)
+				{
+					await _queue.WaitForWork(handler.Kind, Poll, stoppingToken);
+					continue;
+				}
+
+				var host = job.Host ?? string.Empty;
+				inFlight.AddOrUpdate(host, 1, (_, count) => count + 1);
+				try
+				{
+					JobOutcome outcome;
+					try
+					{
+						outcome = await handler.Handle(job, stoppingToken);
+					}
+					catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+					{
+						return;
+					}
+					catch (Exception ex)
+					{
+						_logger.LogError(ex, "{Kind} job {Id} threw", handler.Kind, job.ID);
+						outcome = JobOutcome.Retry(ex.GetType().Name);
+					}
+
+					if (outcome.Result != JobResult.Done && job.Attempts >= handler.MaxAttempts && outcome.Result == JobResult.Retry)
+						_logger.LogWarning("{Kind} job {Id} for {Host} is dead after {Attempts} attempts: {Error}",
+							handler.Kind, job.ID, job.Host, job.Attempts, outcome.Error);
+					await _queue.Finish(job, outcome, handler.MaxAttempts, CancellationToken.None);
+				}
+				catch (Exception ex)
+				{
+					_logger.LogError(ex, "{Kind} job {Id} could not be finished; its lease will expire", handler.Kind, job.ID);
+				}
+				finally
+				{
+					inFlight.AddOrUpdate(host, 0, (_, count) => count - 1);
+				}
+			}
+		}
+
+		async Task Reap(CancellationToken stoppingToken)
+		{
+			while (!stoppingToken.IsCancellationRequested)
+			{
+				try
+				{
+					var reaped = await _queue.Reap(stoppingToken);
+					if (reaped > 0)
+						_logger.LogWarning("{Worker} returned {Count} expired leases to the queue", nameof(JobWorker), reaped);
+				}
+				catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+				{
+					return;
+				}
+				catch (Exception ex)
+				{
+					_logger.LogError(ex, "{Worker} reaper failed", nameof(JobWorker));
+				}
+				await Delay(ReapInterval, stoppingToken);
+			}
+		}
+
+		static async Task Delay(TimeSpan delay, CancellationToken token)
+		{
+			try
+			{
+				await Task.Delay(delay, token);
+			}
+			catch (OperationCanceledException)
+			{
+			}
+		}
+	}
+}

PrivaPub/Middleware/SocialPubConfigurations.cs

@@ -15,6 +15,7 @@ using PrivaPub.Federation.Actors;
 using PrivaPub.Federation.Outbox;
 using PrivaPub.Federation.Inbox;
 using PrivaPub.Infrastructure.Http;
+using PrivaPub.Infrastructure.Jobs;
 using Microsoft.Extensions.Options;
 
 namespace PrivaPub.Middleware
@@ -51,7 +52,10 @@ namespace PrivaPub.Middleware
 				.AddSingleton<IRemoteActorService, RemoteActorService>()
 				.AddSingleton<IDeliveryService, DeliveryService>()
 				.AddTransient<IInboxService, InboxService>()
-				.AddHostedService<DeliveryWorker>();
+				.AddSingleton<IJobQueue, JobQueue>()
+				.AddSingleton<IHostCircuitBreaker, HostCircuitBreaker>()
+				.AddSingleton<IJobHandler, DeliveryJobHandler>()
+				.AddHostedService<JobWorker>();
 		}
 		public static IServiceCollection PrivaPubAuthServicesConfiguration(this IServiceCollection service, IConfiguration configuration)
 		{

PrivaPub/Models/Jobs/Job.cs

@@ -0,0 +1,34 @@
+using MongoDB.Entities;
+
+namespace PrivaPub.Models.Jobs
+{
+	public class Job : Entity
+	{
+		public JobKind Kind { get; set; }
+		public JobState State { get; set; }
+		public string Payload { get; set; }
+		public string Host { get; set; }
+		public string DedupeKey { get; set; }
+		public int Attempts { get; set; }
+		public DateTime RunAt { get; set; } = DateTime.UtcNow;
+		public DateTime? LeasedUntil { get; set; }
+		public string LeaseOwner { get; set; }
+		public string LastError { get; set; }
+		public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
+		public DateTime? FinishedAt { get; set; }
+	}
+
+	public enum JobKind
+	{
+		Deliver,
+		ProcessInbox
+	}
+
+	public enum JobState
+	{
+		Pending,
+		Running,
+		Done,
+		Dead
+	}
+}

PrivaPub/Models/Jobs/RemoteInstance.cs

@@ -0,0 +1,14 @@
+using MongoDB.Entities;
+
+namespace PrivaPub.Models.Jobs
+{
+	public class RemoteInstance : Entity
+	{
+		public string Host { get; set; }
+		public int ConsecutiveFailures { get; set; }
+		public DateTime? UnavailableUntil { get; set; }
+		public DateTime? LastSuccessAt { get; set; }
+		public DateTime? LastFailureAt { get; set; }
+		public string LastError { get; set; }
+	}
+}