SocialPub/PrivaPub.Tests/Infrastructure/JobQueueTests.cs

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<Job>().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<Job>().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<Job>().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<AppConfiguration>(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<DeliveryJobHandler>.Instance);
			using var worker = new JobWorker(queue, new IJobHandler[] { handler }, NullLogger<JobWorker>.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") < 30);
			while (DateTime.UtcNow < deadline && !await DB.Default.Find<RemoteInstance>().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<RemoteInstance>().Match(i => i.Host == "localhost").ExecuteFirstAsync(token);
			Assert.True(instance.ConsecutiveFailures > 0);
		}
	}
}