T1: tests stop sharing state they don't own
- JobQueue takes an optional scope, so a test's worker leases and reaps only its own jobs. - The dead-host delivery test runs alone (Exclusive), on its own jobs, and cleans up the breaker rows it trips; the breaker has tests of its own on unique hosts. - Index and migration tests run alone: they drop indexes and rewrite every post. - DomainBlocks.Load replaces reflection and a database-wide block in tests. - Harness.Outgoing sees only deliveries queued since the harness started: Peer ports are reused within a run, which made the circle test flaky. - Two pure-logic tests leave Mongo-gated classes, so CI runs them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ELjqpznMFMNrJoJUj6K5p2
This commit is contained in:
1 parent
7c6fe80f1b
commit
00b2685cf4
14 files changed
+285
-106
No files matched your search
@@ -0,0 +1,95 @@
|
||||
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
|
||||
{
|
||||
[Xunit.Collection(nameof(Exclusive))]
|
||||
[Trait("Category", "Integration")]
|
||||
public sealed class DeliveryWorkerTests : IAsyncLifetime
|
||||
{
|
||||
static readonly string[] PeerHosts = { "localhost", "127.0.0.1" };
|
||||
|
||||
Peer _peer;
|
||||
|
||||
public async ValueTask InitializeAsync()
|
||||
{
|
||||
Assert.SkipUnless(MongoFixture.Enabled, MongoFixture.Skip);
|
||||
_peer = await Peer.Start();
|
||||
await DB.Default.DeleteAsync<RemoteInstance>(i => PeerHosts.Contains(i.Host));
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_peer == default)
|
||||
return;
|
||||
await _peer.DisposeAsync();
|
||||
await DB.Default.DeleteAsync<RemoteInstance>(i => PeerHosts.Contains(i.Host));
|
||||
}
|
||||
|
||||
[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 run = $"https://privapub.test/a/{Guid.NewGuid():N}/";
|
||||
var queue = new JobQueue(j => j.DedupeKey.StartsWith(run));
|
||||
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"] = run + "dead", ["type"] = "Create" }, token);
|
||||
await delivery.Enqueue(sender, new[] { $"{_peer.A}/live/inbox" }, new JsonObject { ["id"] = run + "live", ["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);
|
||||
try
|
||||
{
|
||||
await Until(() => Task.FromResult(_peer.Requests.Any(r => r.Path == "/live/inbox")), token);
|
||||
var deadSoFar = _peer.Requests.Count(r => r.Path == "/dead/inbox");
|
||||
var live = Assert.Single(_peer.Requests, r => r.Path == "/live/inbox");
|
||||
Assert.Contains($"keyId=\"{sender.KeyId}\"", live.Signature);
|
||||
Assert.True(deadSoFar < 30, $"{deadSoFar} dead deliveries were tried before the live one");
|
||||
|
||||
await Until(() => DB.Default.Find<Job>().Match(j => j.DedupeKey.StartsWith(run + "dead") && j.State == JobState.Pending && j.Attempts > 0).ExecuteAnyAsync(token), token);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await worker.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
var retried = await DB.Default.Find<Job>().Match(j => j.DedupeKey.StartsWith(run + "dead") && j.Attempts > 0 && j.State == JobState.Pending).ExecuteFirstAsync(token);
|
||||
Assert.Equal("503", retried.LastError?.Split(' ')[0]);
|
||||
Assert.True(retried.RunAt > DateTime.UtcNow);
|
||||
Assert.True((await DB.Default.Find<RemoteInstance>().Match(i => i.Host == "localhost").ExecuteFirstAsync(token))?.ConsecutiveFailures > 0);
|
||||
}
|
||||
|
||||
static async Task Until(Func<Task<bool>> condition, CancellationToken token)
|
||||
{
|
||||
var deadline = DateTime.UtcNow.AddSeconds(30);
|
||||
while (!await condition())
|
||||
{
|
||||
Assert.True(DateTime.UtcNow < deadline, "timed out");
|
||||
await Task.Delay(50, token);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
using Microsoft.Extensions.Caching.Memory;
|
||||
|
||||
using MongoDB.Entities;
|
||||
|
||||
using PrivaPub.Infrastructure.Jobs;
|
||||
using PrivaPub.Models.Jobs;
|
||||
using PrivaPub.Tests.Support;
|
||||
|
||||
namespace PrivaPub.Tests.Infrastructure
|
||||
{
|
||||
[Trait("Category", "Integration")]
|
||||
public sealed class HostCircuitBreakerTests : IAsyncLifetime
|
||||
{
|
||||
readonly string _host = $"breaker{Guid.NewGuid():N}.example";
|
||||
|
||||
public ValueTask InitializeAsync()
|
||||
{
|
||||
Assert.SkipUnless(MongoFixture.Enabled, MongoFixture.Skip);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (MongoFixture.Enabled)
|
||||
await DB.Default.DeleteAsync<RemoteInstance>(i => i.Host == _host);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_host_is_quarantined_only_past_the_threshold_and_freed_by_a_success()
|
||||
{
|
||||
var token = TestContext.Current.CancellationToken;
|
||||
var breaker = new HostCircuitBreaker(new MemoryCache(new MemoryCacheOptions()));
|
||||
|
||||
for (var i = 1; i < HostCircuitBreaker.Threshold; i++)
|
||||
await breaker.Failed(_host, "503", token);
|
||||
Assert.Null(await breaker.UnavailableUntil(_host, token));
|
||||
|
||||
await breaker.Failed(_host, "503", token);
|
||||
var until = await breaker.UnavailableUntil(_host, token);
|
||||
Assert.NotNull(until);
|
||||
Assert.InRange(until.Value, DateTime.UtcNow.AddMinutes(55), DateTime.UtcNow.AddMinutes(65));
|
||||
var instance = await DB.Default.Find<RemoteInstance>().Match(i => i.Host == _host).ExecuteFirstAsync(token);
|
||||
Assert.Equal(HostCircuitBreaker.Threshold, instance.ConsecutiveFailures);
|
||||
Assert.Equal("503", instance.LastError);
|
||||
|
||||
await breaker.Succeeded(_host, token);
|
||||
Assert.Null(await breaker.UnavailableUntil(_host, token));
|
||||
instance = await DB.Default.Find<RemoteInstance>().Match(i => i.Host == _host).ExecuteFirstAsync(token);
|
||||
Assert.Equal(0, instance.ConsecutiveFailures);
|
||||
Assert.NotNull(instance.LastSuccessAt);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_healthy_host_is_written_at_most_once_an_hour()
|
||||
{
|
||||
var token = TestContext.Current.CancellationToken;
|
||||
var breaker = new HostCircuitBreaker(new MemoryCache(new MemoryCacheOptions()));
|
||||
|
||||
await breaker.Succeeded(_host, token);
|
||||
var first = (await DB.Default.Find<RemoteInstance>().Match(i => i.Host == _host).ExecuteFirstAsync(token)).LastSuccessAt;
|
||||
await breaker.Succeeded(_host, token);
|
||||
var second = (await DB.Default.Find<RemoteInstance>().Match(i => i.Host == _host).ExecuteFirstAsync(token)).LastSuccessAt;
|
||||
|
||||
Assert.Equal(first, second);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -12,6 +12,7 @@ using PrivaPub.Tests.Support;
|
||||
|
||||
namespace PrivaPub.Tests.Infrastructure
|
||||
{
|
||||
[Xunit.Collection(nameof(Exclusive))]
|
||||
[Trait("Category", "Integration")]
|
||||
public class IndexTests
|
||||
{
|
||||
|
||||
@@ -1,19 +1,9 @@
|
||||
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
|
||||
@@ -38,19 +28,13 @@ namespace PrivaPub.Tests.Infrastructure
|
||||
[Trait("Category", "Integration")]
|
||||
public sealed class JobQueueTests : IAsyncLifetime
|
||||
{
|
||||
Peer _peer;
|
||||
|
||||
public async ValueTask InitializeAsync()
|
||||
public ValueTask InitializeAsync()
|
||||
{
|
||||
Assert.SkipUnless(MongoFixture.Enabled, MongoFixture.Skip);
|
||||
_peer = await Peer.Start();
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_peer != default)
|
||||
await _peer.DisposeAsync();
|
||||
}
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
|
||||
static Job NewJob(string host = default, string dedupe = default) => new()
|
||||
{
|
||||
@@ -102,44 +86,9 @@ namespace PrivaPub.Tests.Infrastructure
|
||||
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);
|
||||
await new JobQueue(j => j.ID == job.ID).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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,7 @@ using PrivaPub.Tests.Support;
|
||||
|
||||
namespace PrivaPub.Tests.Infrastructure
|
||||
{
|
||||
[Xunit.Collection(nameof(Exclusive))]
|
||||
[Trait("Category", "Integration")]
|
||||
public class MigrationTests
|
||||
{
|
||||
|
||||
Reference in new issue
Block a user