From 8f5fc36e675bf8e42b9b97ca0a6e456e24b610d8 Mon Sep 17 00:00:00 2001 From: Samuel McAravey Date: Thu, 18 Dec 2025 22:10:04 -0800 Subject: [PATCH 1/2] Add fanout coordinator integration coverage --- .../FanoutCoordinatorIntegrationTests.cs | 376 ++++++++++++++++++ 1 file changed, 376 insertions(+) create mode 100644 tests/Bravellian.Platform.Tests/FanoutCoordinatorIntegrationTests.cs diff --git a/tests/Bravellian.Platform.Tests/FanoutCoordinatorIntegrationTests.cs b/tests/Bravellian.Platform.Tests/FanoutCoordinatorIntegrationTests.cs new file mode 100644 index 00000000..cb93d7ff --- /dev/null +++ b/tests/Bravellian.Platform.Tests/FanoutCoordinatorIntegrationTests.cs @@ -0,0 +1,376 @@ +// Copyright (c) Bravellian +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +using System.Collections.Concurrent; +using System.Runtime.CompilerServices; +using Bravellian.Platform.Outbox; +using Dapper; +using Microsoft.Data.SqlClient; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Shouldly; + +namespace Bravellian.Platform.Tests; + +[Collection(SqlServerCollection.Name)] +[Trait("Category", "Integration")] +[Trait("RequiresDocker", "true")] +public class FanoutCoordinatorIntegrationTests : SqlServerTestBase +{ + private SqlOutboxOptions? outboxOptions; + private SqlFanoutOptions? fanoutOptions; + private SqlOutboxJoinStore? joinStore; + private SqlOutboxService? outboxService; + + public FanoutCoordinatorIntegrationTests(ITestOutputHelper testOutputHelper, SqlServerCollectionFixture fixture) + : base(testOutputHelper, fixture) + { + } + + public override async ValueTask InitializeAsync() + { + await base.InitializeAsync().ConfigureAwait(false); + + outboxOptions = new SqlOutboxOptions + { + ConnectionString = ConnectionString, + SchemaName = "dbo", + TableName = "Outbox", + }; + + fanoutOptions = new SqlFanoutOptions + { + ConnectionString = ConnectionString, + SchemaName = "dbo", + PolicyTableName = "FanoutPolicy", + CursorTableName = "FanoutCursor", + }; + + await DatabaseSchemaManager.EnsureFanoutSchemaAsync( + ConnectionString, + fanoutOptions.SchemaName, + fanoutOptions.PolicyTableName, + fanoutOptions.CursorTableName).ConfigureAwait(false); + + joinStore = new SqlOutboxJoinStore( + Options.Create(outboxOptions), + NullLogger.Instance); + + outboxService = new SqlOutboxService( + Options.Create(outboxOptions), + NullLogger.Instance, + joinStore); + } + + [Fact] + public async Task RunAsync_RespectsActiveLeaseAndRecoversAfterExpiry() + { + var leaseFactory = new InMemoryLeaseFactory(TimeSpan.FromMilliseconds(200)); + await using var heldLease = await leaseFactory.AcquireAsync( + "fanout:billing", + TimeSpan.FromMinutes(1), + cancellationToken: TestContext.Current.CancellationToken); + + var planner = new StaticPlanner(new[] { new FanoutSlice("billing", "tenant-1", "daily") }); + var dispatcher = new FanoutDispatcher(outboxService!); + var coordinator = new FanoutCoordinator(planner, dispatcher, leaseFactory, NullLogger.Instance); + + var blockedResult = await coordinator.RunAsync("billing", null, TestContext.Current.CancellationToken); + blockedResult.ShouldBe(0); + + await Task.Delay(400, TestContext.Current.CancellationToken); + + var dispatched = await coordinator.RunAsync("billing", null, TestContext.Current.CancellationToken); + dispatched.ShouldBe(1); + + var messageCount = await CountOutboxMessagesAsync(); + messageCount.ShouldBe(1); + } + + [Fact] + public async Task RunAsync_RedispatchesAbandonedSlices() + { + var leaseFactory = new InMemoryLeaseFactory(); + var slice = new FanoutSlice("analytics", "tenant-7", "hourly"); + var planner = new StaticPlanner(new[] { slice }); + var dispatcher = new FanoutDispatcher(outboxService!); + var coordinator = new FanoutCoordinator(planner, dispatcher, leaseFactory, NullLogger.Instance); + + var firstPass = await coordinator.RunAsync("analytics", null, TestContext.Current.CancellationToken); + firstPass.ShouldBe(1); + + var secondPass = await coordinator.RunAsync("analytics", null, TestContext.Current.CancellationToken); + secondPass.ShouldBe(1); + + var count = await CountOutboxMessagesAsync(); + count.ShouldBe(2); + + var payloads = await GetOutboxPayloadsAsync(); + payloads.ShouldAllBe(p => p.Contains("tenant-7", StringComparison.OrdinalIgnoreCase)); + } + + [Fact] + public async Task RunAsync_SkipsCompletedSlicesAfterCursorAdvances() + { + var leaseFactory = new InMemoryLeaseFactory(); + var policyRepo = new SqlFanoutPolicyRepository(Options.Create(fanoutOptions!)); + var cursorRepo = new SqlFanoutCursorRepository(Options.Create(fanoutOptions!)); + + await policyRepo.SetCadenceAsync("reports", "daily", everySeconds: 30, jitterSeconds: 0, TestContext.Current.CancellationToken); + + var planner = new ShardedPlanner(policyRepo, cursorRepo, TimeProvider.System, ["shard-a", "shard-b"]); + var dispatcher = new FanoutDispatcher(outboxService!); + var coordinator = new FanoutCoordinator(planner, dispatcher, leaseFactory, NullLogger.Instance); + + var initialDispatch = await coordinator.RunAsync("reports", null, TestContext.Current.CancellationToken); + initialDispatch.ShouldBe(2); + + var completionTime = DateTimeOffset.UtcNow; + await Task.WhenAll( + cursorRepo.MarkCompletedAsync("reports", "daily", "shard-a", completionTime, TestContext.Current.CancellationToken), + cursorRepo.MarkCompletedAsync("reports", "daily", "shard-b", completionTime, TestContext.Current.CancellationToken)); + + var afterCompletion = await coordinator.RunAsync("reports", null, TestContext.Current.CancellationToken); + afterCompletion.ShouldBe(0); + } + + [Fact] + public async Task FanoutSlices_CanJoinDownstreamMessagesIdempotently() + { + var leaseFactory = new InMemoryLeaseFactory(); + var dispatcher = new FanoutDispatcher(outboxService!); + var correlationId = Guid.NewGuid().ToString("N"); + + var slices = new[] + { + new FanoutSlice("fanout", "a", "work", correlationId: correlationId), + new FanoutSlice("fanout", "b", "work", correlationId: correlationId), + new FanoutSlice("fanout", "c", "work", correlationId: correlationId), + }; + + var planner = new StaticPlanner(slices); + var coordinator = new FanoutCoordinator(planner, dispatcher, leaseFactory, NullLogger.Instance); + + var dispatched = await coordinator.RunAsync("fanout", null, TestContext.Current.CancellationToken); + dispatched.ShouldBe(3); + + var join = await joinStore!.CreateJoinAsync(42, expectedSteps: 3, metadata: "fan-in", TestContext.Current.CancellationToken); + + var outboxMessages = await GetOutboxMessagesAsync(); + foreach (var message in outboxMessages) + { + await joinStore.AttachMessageToJoinAsync(join.JoinId, OutboxMessageIdentifier.From(message.Id), TestContext.Current.CancellationToken); + } + + var unorderedIds = outboxMessages.Select(m => OutboxMessageIdentifier.From(m.Id)).Reverse().ToArray(); + foreach (var id in unorderedIds) + { + await joinStore.IncrementCompletedAsync(join.JoinId, id, TestContext.Current.CancellationToken); + } + + var finalJoin = await joinStore.GetJoinAsync(join.JoinId, TestContext.Current.CancellationToken); + finalJoin!.CompletedSteps.ShouldBe(3); + finalJoin.ExpectedSteps.ShouldBe(3); + + // Replaying the last slice should not double-count + await joinStore.IncrementCompletedAsync(join.JoinId, unorderedIds.Last(), TestContext.Current.CancellationToken); + var replayed = await joinStore.GetJoinAsync(join.JoinId, TestContext.Current.CancellationToken); + replayed!.CompletedSteps.ShouldBe(3); + + var correlationIds = outboxMessages.Select(m => m.CorrelationId).Distinct().ToList(); + correlationIds.Count.ShouldBe(1); + } + + private async Task CountOutboxMessagesAsync() + { + await using var connection = new SqlConnection(ConnectionString); + await connection.OpenAsync(TestContext.Current.CancellationToken); + var count = await connection.ExecuteScalarAsync( + new CommandDefinition("SELECT COUNT(*) FROM dbo.Outbox", cancellationToken: TestContext.Current.CancellationToken)); + return count; + } + + private async Task> GetOutboxPayloadsAsync() + { + await using var connection = new SqlConnection(ConnectionString); + await connection.OpenAsync(TestContext.Current.CancellationToken); + var payloads = await connection.QueryAsync( + new CommandDefinition("SELECT Payload FROM dbo.Outbox ORDER BY CreatedAt", cancellationToken: TestContext.Current.CancellationToken)); + return payloads.ToList(); + } + + private async Task> GetOutboxMessagesAsync() + { + await using var connection = new SqlConnection(ConnectionString); + await connection.OpenAsync(TestContext.Current.CancellationToken); + var messages = await connection.QueryAsync<(Guid Id, string CorrelationId)>( + new CommandDefinition("SELECT Id, CorrelationId FROM dbo.Outbox ORDER BY CreatedAt", cancellationToken: TestContext.Current.CancellationToken)); + return messages.ToList(); + } + + private sealed class StaticPlanner : IFanoutPlanner + { + private readonly IReadOnlyList slices; + + public StaticPlanner(IEnumerable slices) + { + this.slices = slices.ToList(); + } + + public Task> GetDueSlicesAsync(string fanoutTopic, string? workKey, CancellationToken ct) + { + return Task.FromResult(slices); + } + } + + private sealed class ShardedPlanner : BaseFanoutPlanner + { + private readonly IReadOnlyList shards; + + public ShardedPlanner( + IFanoutPolicyRepository policyRepository, + IFanoutCursorRepository cursorRepository, + TimeProvider timeProvider, + IReadOnlyList shards) + : base(policyRepository, cursorRepository, timeProvider) + { + this.shards = shards; + } + + protected override async IAsyncEnumerable<(string ShardKey, string WorkKey)> EnumerateCandidatesAsync( + string fanoutTopic, + string? workKey, + [EnumeratorCancellation] CancellationToken ct) + { + var wk = workKey ?? "default"; + foreach (var shard in shards) + { + yield return (shard, wk); + await Task.Yield(); + } + } + } + + private sealed class InMemoryLeaseFactory : ISystemLeaseFactory + { + private readonly ConcurrentDictionary leases = new(StringComparer.Ordinal); + private readonly TimeSpan? overrideDuration; + private long fencingToken; + + public InMemoryLeaseFactory(TimeSpan? overrideDuration = null) + { + this.overrideDuration = overrideDuration; + } + + public Task AcquireAsync( + string resourceName, + TimeSpan duration, + string? contextJson = null, + OwnerToken? ownerToken = default, + CancellationToken cancellationToken = default) + { + var effectiveDuration = overrideDuration ?? duration; + + while (leases.TryGetValue(resourceName, out var existing)) + { + if (existing.IsExpired) + { + leases.TryRemove(resourceName, out _); + } + else + { + return Task.FromResult(null); + } + } + + var nextToken = Interlocked.Increment(ref fencingToken); + var lease = new InMemoryLease(resourceName, nextToken, effectiveDuration, ownerToken ?? OwnerToken.GenerateNew(), RemoveLease); + leases[resourceName] = lease; + return Task.FromResult(lease); + } + + private void RemoveLease(string resourceName) + { + leases.TryRemove(resourceName, out _); + } + + private sealed class InMemoryLease : ISystemLease + { + private readonly CancellationTokenSource cts; + private readonly Action onDispose; + private readonly TimeSpan duration; + private readonly DateTimeOffset acquired; + + public InMemoryLease(string resourceName, long fencingToken, TimeSpan duration, OwnerToken ownerToken, Action onDispose) + { + ResourceName = resourceName; + FencingToken = fencingToken; + OwnerToken = ownerToken; + this.duration = duration; + this.onDispose = onDispose; + acquired = DateTimeOffset.UtcNow; + cts = new CancellationTokenSource(); + _ = Task.Run(async () => + { + try + { + await Task.Delay(duration, CancellationToken.None).ConfigureAwait(false); + cts.Cancel(); + onDispose(resourceName); + } + catch + { + } + }); + } + + public string ResourceName { get; } + + public OwnerToken OwnerToken { get; } + + public long FencingToken { get; private set; } + + public CancellationToken CancellationToken => cts.Token; + + public bool IsExpired => DateTimeOffset.UtcNow - acquired >= duration; + + public void ThrowIfLost() + { + if (IsExpired) + { + throw new LostLeaseException(ResourceName, OwnerToken); + } + } + + public Task TryRenewNowAsync(CancellationToken cancellationToken = default) + { + if (IsExpired) + { + return Task.FromResult(false); + } + + FencingToken++; + return Task.FromResult(true); + } + + public ValueTask DisposeAsync() + { + cts.Cancel(); + cts.Dispose(); + onDispose(ResourceName); + return ValueTask.CompletedTask; + } + } + } +} From a6e4473d5b8c119f94f4c243bad17fca28acd1e8 Mon Sep 17 00:00:00 2001 From: Copilot <198982749+Copilot@users.noreply.github.com> Date: Thu, 18 Dec 2025 22:25:53 -0800 Subject: [PATCH 2/2] Fix OutboxJoin schema initialization and InMemoryLease concurrency issues in fanout tests (#99) Co-authored-by: SamuelMcAravey <11021165+SamuelMcAravey@users.noreply.github.com> Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> --- .../FanoutCoordinatorIntegrationTests.cs | 74 +++++++++++++------ 1 file changed, 52 insertions(+), 22 deletions(-) diff --git a/tests/Bravellian.Platform.Tests/FanoutCoordinatorIntegrationTests.cs b/tests/Bravellian.Platform.Tests/FanoutCoordinatorIntegrationTests.cs index cb93d7ff..ce1a1fdb 100644 --- a/tests/Bravellian.Platform.Tests/FanoutCoordinatorIntegrationTests.cs +++ b/tests/Bravellian.Platform.Tests/FanoutCoordinatorIntegrationTests.cs @@ -63,6 +63,10 @@ await DatabaseSchemaManager.EnsureFanoutSchemaAsync( fanoutOptions.PolicyTableName, fanoutOptions.CursorTableName).ConfigureAwait(false); + await DatabaseSchemaManager.EnsureOutboxJoinSchemaAsync( + ConnectionString, + outboxOptions.SchemaName).ConfigureAwait(false); + joinStore = new SqlOutboxJoinStore( Options.Create(outboxOptions), NullLogger.Instance); @@ -127,7 +131,7 @@ public async Task RunAsync_SkipsCompletedSlicesAfterCursorAdvances() var policyRepo = new SqlFanoutPolicyRepository(Options.Create(fanoutOptions!)); var cursorRepo = new SqlFanoutCursorRepository(Options.Create(fanoutOptions!)); - await policyRepo.SetCadenceAsync("reports", "daily", everySeconds: 30, jitterSeconds: 0, TestContext.Current.CancellationToken); + await policyRepo.SetCadenceAsync("reports", "default", everySeconds: 30, jitterSeconds: 0, TestContext.Current.CancellationToken); var planner = new ShardedPlanner(policyRepo, cursorRepo, TimeProvider.System, ["shard-a", "shard-b"]); var dispatcher = new FanoutDispatcher(outboxService!); @@ -138,8 +142,8 @@ public async Task RunAsync_SkipsCompletedSlicesAfterCursorAdvances() var completionTime = DateTimeOffset.UtcNow; await Task.WhenAll( - cursorRepo.MarkCompletedAsync("reports", "daily", "shard-a", completionTime, TestContext.Current.CancellationToken), - cursorRepo.MarkCompletedAsync("reports", "daily", "shard-b", completionTime, TestContext.Current.CancellationToken)); + cursorRepo.MarkCompletedAsync("reports", "default", "shard-a", completionTime, TestContext.Current.CancellationToken), + cursorRepo.MarkCompletedAsync("reports", "default", "shard-b", completionTime, TestContext.Current.CancellationToken)); var afterCompletion = await coordinator.RunAsync("reports", null, TestContext.Current.CancellationToken); afterCompletion.ShouldBe(0); @@ -282,22 +286,37 @@ public InMemoryLeaseFactory(TimeSpan? overrideDuration = null) { var effectiveDuration = overrideDuration ?? duration; - while (leases.TryGetValue(resourceName, out var existing)) + while (true) { - if (existing.IsExpired) + var nextToken = Interlocked.Increment(ref fencingToken); + var newLease = new InMemoryLease( + resourceName, + nextToken, + effectiveDuration, + ownerToken ?? OwnerToken.GenerateNew(), + RemoveLease); + + var resultingLease = leases.AddOrUpdate( + resourceName, + _ => newLease, + (_, existing) => existing.IsExpired ? newLease : existing); + + if (ReferenceEquals(resultingLease, newLease)) { - leases.TryRemove(resourceName, out _); + // We successfully installed our lease. + return Task.FromResult(newLease); } - else + + if (!resultingLease.IsExpired) { + // Another thread holds a non-expired lease. + newLease.DisposeAsync().AsTask().Wait(); return Task.FromResult(null); } - } - var nextToken = Interlocked.Increment(ref fencingToken); - var lease = new InMemoryLease(resourceName, nextToken, effectiveDuration, ownerToken ?? OwnerToken.GenerateNew(), RemoveLease); - leases[resourceName] = lease; - return Task.FromResult(lease); + // If the resulting lease is expired, dispose the unused lease and loop to try again. + newLease.DisposeAsync().AsTask().Wait(); + } } private void RemoveLease(string resourceName) @@ -311,26 +330,29 @@ private sealed class InMemoryLease : ISystemLease private readonly Action onDispose; private readonly TimeSpan duration; private readonly DateTimeOffset acquired; + private readonly Task expirationTask; + private long fencingToken; - public InMemoryLease(string resourceName, long fencingToken, TimeSpan duration, OwnerToken ownerToken, Action onDispose) + public InMemoryLease(string resourceName, long initialFencingToken, TimeSpan duration, OwnerToken ownerToken, Action onDispose) { ResourceName = resourceName; - FencingToken = fencingToken; + fencingToken = initialFencingToken; OwnerToken = ownerToken; this.duration = duration; this.onDispose = onDispose; acquired = DateTimeOffset.UtcNow; cts = new CancellationTokenSource(); - _ = Task.Run(async () => + expirationTask = Task.Run(async () => { try { - await Task.Delay(duration, CancellationToken.None).ConfigureAwait(false); - cts.Cancel(); + await Task.Delay(duration, cts.Token).ConfigureAwait(false); + // If we reach here, the delay completed without cancellation, so the lease expired onDispose(resourceName); } - catch + catch (OperationCanceledException) { + // Expected when lease is disposed before expiration } }); } @@ -339,7 +361,7 @@ public InMemoryLease(string resourceName, long fencingToken, TimeSpan duration, public OwnerToken OwnerToken { get; } - public long FencingToken { get; private set; } + public long FencingToken => Interlocked.Read(ref fencingToken); public CancellationToken CancellationToken => cts.Token; @@ -360,16 +382,24 @@ public Task TryRenewNowAsync(CancellationToken cancellationToken = default return Task.FromResult(false); } - FencingToken++; + Interlocked.Increment(ref fencingToken); return Task.FromResult(true); } - public ValueTask DisposeAsync() + public async ValueTask DisposeAsync() { cts.Cancel(); + try + { + await expirationTask.ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Expected when disposing before expiration + } + cts.Dispose(); onDispose(ResourceName); - return ValueTask.CompletedTask; } } }