using Microsoft.Extensions.Caching.Memory; using Microsoft.Extensions.Logging.Abstractions; using MongoDB.Entities; using PrivaPub.Federation.Actors; using PrivaPub.Federation.Outbox; using PrivaPub.Infrastructure.Jobs; using PrivaPub.Models; using PrivaPub.Models.Jobs; using PrivaPub.Models.User; using PrivaPub.StaticServices; using PrivaPub.Tests.Support; using System.Text.Json.Nodes; namespace PrivaPub.Tests.Infrastructure { public class BackoffTests { [Fact] public void Grows_like_mastodons() { Assert.InRange(Backoff.After(1).TotalSeconds, 16, 16 + 9 * 2); Assert.InRange(Backoff.After(10).TotalSeconds, 10015, 10015 + 9 * 11); } [Fact] public void Quarantines_a_host_only_past_the_threshold() { Assert.Equal(TimeSpan.Zero, Backoff.HostQuarantine(9, 10)); Assert.Equal(TimeSpan.FromHours(1), Backoff.HostQuarantine(10, 10)); Assert.Equal(TimeSpan.FromHours(4), Backoff.HostQuarantine(12, 10)); Assert.Equal(TimeSpan.FromDays(7), Backoff.HostQuarantine(40, 10)); } } [Trait("Category", "Integration")] public sealed class JobQueueTests : IAsyncLifetime { Peer _peer; public async ValueTask InitializeAsync() { Assert.SkipUnless(MongoFixture.Enabled, MongoFixture.Skip); _peer = await Peer.Start(); } public async ValueTask DisposeAsync() { if (_peer != default) await _peer.DisposeAsync(); } static Job NewJob(string host = default, string dedupe = default) => new() { Kind = (JobKind)99, Host = host, DedupeKey = dedupe, Payload = "{}" }; [Fact] public async Task A_dedupe_key_is_queued_once() { var queue = new JobQueue(); var key = Guid.NewGuid().ToString("N"); var inserted = await queue.EnqueueMany(new[] { NewJob(dedupe: key), NewJob(dedupe: key) }, TestContext.Current.CancellationToken); Assert.Equal(1, inserted); } [Fact] public async Task A_busy_host_is_skipped_and_a_failed_job_backs_off() { var token = TestContext.Current.CancellationToken; var queue = new JobQueue(); var kind = (JobKind)(100 + Random.Shared.Next(1000)); var busy = $"busy{Guid.NewGuid():N}.example"; var free = $"free{Guid.NewGuid():N}.example"; await queue.EnqueueMany(new[] { new Job { Kind = kind, Host = busy }, new Job { Kind = kind, Host = free } }, token); var leased = await queue.Lease(kind, new[] { busy }, token); Assert.Equal(free, leased.Host); Assert.Equal(1, leased.Attempts); await queue.Finish(leased, JobOutcome.Retry("boom"), maxAttempts: 3, token); var after = await DB.Default.Find().OneAsync(leased.ID, token); Assert.Equal(JobState.Pending, after.State); Assert.True(after.RunAt > DateTime.UtcNow.AddSeconds(10)); leased.Attempts = 3; await queue.Finish(leased, JobOutcome.Retry("boom"), maxAttempts: 3, token); Assert.Equal(JobState.Dead, (await DB.Default.Find().OneAsync(leased.ID, token)).State); } [Fact] public async Task An_expired_lease_returns_to_the_queue() { var token = TestContext.Current.CancellationToken; var job = new Job { Kind = (JobKind)98, State = JobState.Running, LeasedUntil = DateTime.UtcNow.AddMinutes(-1) }; await DB.Default.SaveAsync(job, token); await new JobQueue().Reap(token); Assert.Equal(JobState.Pending, (await DB.Default.Find().OneAsync(job.ID, token)).State); } [Fact] public async Task A_dead_host_does_not_hold_up_deliveries_to_live_ones() { var token = TestContext.Current.CancellationToken; var (privateKey, publicKey) = Keys.NewKeyPair(); var avatar = new Avatar { UserName = $"sender{Guid.NewGuid():N}"[..20], PrivateKey = privateKey, PublicKey = publicKey }; await DB.Default.SaveAsync(avatar, token); var local = new LocalActorService(new DbEntities(), new StaticOptions(new AppConfiguration { BackendBaseAddress = "https://privapub.test" })); var sender = local.FromAvatar(avatar); var queue = new JobQueue(); var delivery = new DeliveryService(new DbEntities(), queue); _peer.Answer("/dead/inbox", 503, TimeSpan.FromMilliseconds(300)); _peer.Answer("/live/inbox", 202); var deadInboxes = Enumerable.Range(0, 60).Select(i => $"{_peer.B}/dead/inbox?{i}"); await delivery.Enqueue(sender, deadInboxes, new JsonObject { ["id"] = $"https://privapub.test/a/{Guid.NewGuid():N}", ["type"] = "Create" }, token); await delivery.Enqueue(sender, new[] { $"{_peer.A}/live/inbox" }, new JsonObject { ["id"] = $"https://privapub.test/a/{Guid.NewGuid():N}", ["type"] = "Create" }, token); var handler = new DeliveryJobHandler(local, Peer.Http(), new HostCircuitBreaker(new MemoryCache(new MemoryCacheOptions())), NullLogger.Instance); using var worker = new JobWorker(queue, new IJobHandler[] { handler }, NullLogger.Instance); await worker.StartAsync(token); var deadline = DateTime.UtcNow.AddSeconds(5); while (DateTime.UtcNow < deadline && !_peer.Requests.Any(r => r.Path == "/live/inbox")) await Task.Delay(50, token); var live = Assert.Single(_peer.Requests, r => r.Path == "/live/inbox"); Assert.True(_peer.Requests.Count(r => r.Path == "/dead/inbox") < 10); while (DateTime.UtcNow < deadline && !await DB.Default.Find().Match(i => i.Host == "localhost").ExecuteAnyAsync(token)) await Task.Delay(50, token); await worker.StopAsync(token); Assert.Contains($"keyId=\"{sender.KeyId}\"", live.Signature); var instance = await DB.Default.Find().Match(i => i.Host == "localhost").ExecuteFirstAsync(token); Assert.True(instance.ConsecutiveFailures > 0); } } }