Add fanout coordinator integration coverage - #98
Conversation
There was a problem hiding this comment.
Pull request overview
This PR adds comprehensive integration tests for the FanoutCoordinator component, covering lease acquisition, slice redispatch, cursor advancement, and fan-in join operations. The tests use in-memory test doubles to simulate distributed coordination scenarios without requiring external infrastructure.
- Introduces four integration test scenarios validating fanout coordinator behavior
- Implements reusable test helper classes (StaticPlanner, ShardedPlanner, InMemoryLeaseFactory) for testing fanout scenarios
- Adds helper methods for querying outbox message state during tests
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| await DatabaseSchemaManager.EnsureFanoutSchemaAsync( | ||
| ConnectionString, | ||
| fanoutOptions.SchemaName, | ||
| fanoutOptions.PolicyTableName, | ||
| fanoutOptions.CursorTableName).ConfigureAwait(false); | ||
|
|
||
| joinStore = new SqlOutboxJoinStore( | ||
| Options.Create(outboxOptions), | ||
| NullLogger<SqlOutboxJoinStore>.Instance); | ||
|
|
||
| outboxService = new SqlOutboxService( |
There was a problem hiding this comment.
InitializeAsync provisions the fanout schema but never creates the OutboxJoin/OutboxJoinMember tables before instantiating SqlOutboxJoinStore and SqlOutboxService. SqlServerTestBase only installs the Outbox tables, so when the FanoutSlices_CanJoinDownstreamMessagesIdempotently test later calls joinStore.CreateJoinAsync the queries will fail with Invalid object name 'dbo.OutboxJoin' because those tables were never created. Call DatabaseSchemaManager.EnsureOutboxJoinSchemaAsync here before constructing the join store to make the integration test runnable.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Ensure outbox join schema before using join store
InitializeAsync provisions the fanout schema but never creates the
OutboxJoin/OutboxJoinMembertables before instantiatingSqlOutboxJoinStoreandSqlOutboxService. SqlServerTestBase only installs the Outbox tables, so when theFanoutSlices_CanJoinDownstreamMessagesIdempotentlytest later callsjoinStore.CreateJoinAsyncthe queries will fail withInvalid object name 'dbo.OutboxJoin'because those tables were never created. CallDatabaseSchemaManager.EnsureOutboxJoinSchemaAsynchere before constructing the join store to make the integration test runnable.Useful? React with 👍 / 👎.
|
@copilot I'm getting the following error in the tests. Please fix: Test run for /home/runner/work/platform/platform/tests/Bravellian.Platform.Tests/bin/Release/net10.0/Bravellian.Platform.Tests.dll (.NETCoreApp,Version=v10.0) |
|
@SamuelMcAravey I've opened a new pull request, #99, to work on those changes. Once the pull request is ready, I'll request review from you. |
…sues 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>
|
@SamuelMcAravey I've opened a new pull request, #100, to work on those changes. Once the pull request is ready, I'll request review from you. |
|
InitializeAsync provisions the fanout schema but never creates the |
|
@SamuelMcAravey I've opened a new pull request, #101, to work on those changes. Once the pull request is ready, I'll request review from you. |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 1 out of 1 changed files in this pull request and generated 8 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| foreach (var shard in shards) | ||
| { | ||
| yield return (shard, wk); | ||
| await Task.Yield(); |
There was a problem hiding this comment.
The Task.Yield() here appears unnecessary. In test code, yielding control after each iteration adds overhead without providing value. The async enumerable pattern already provides sufficient asynchrony without explicit yields in simple enumeration scenarios.
| await Task.Yield(); |
| public async ValueTask DisposeAsync() | ||
| { | ||
| cts.Cancel(); | ||
| try | ||
| { | ||
| await expirationTask.ConfigureAwait(false); | ||
| } | ||
| catch (OperationCanceledException) | ||
| { | ||
| // Expected when disposing before expiration | ||
| } | ||
|
|
||
| cts.Dispose(); | ||
| onDispose(ResourceName); | ||
| } |
There was a problem hiding this comment.
There is a race condition between disposing the lease and the expiration callback. When DisposeAsync is called, it cancels the token and waits for expirationTask. However, onDispose is called both when the lease expires naturally (line 351) and when it's disposed manually (line 402). If disposal happens right as the lease is expiring, onDispose could be called twice concurrently, potentially causing issues with the ConcurrentDictionary removal in the parent class.
|
|
||
| private sealed class InMemoryLeaseFactory : ISystemLeaseFactory | ||
| { | ||
| private readonly ConcurrentDictionary<string, InMemoryLease> leases = new(StringComparer.Ordinal); | ||
| private readonly TimeSpan? overrideDuration; | ||
| private long fencingToken; | ||
|
|
||
| public InMemoryLeaseFactory(TimeSpan? overrideDuration = null) | ||
| { | ||
| this.overrideDuration = overrideDuration; | ||
| } | ||
|
|
There was a problem hiding this comment.
This test helper class lacks documentation. Consider adding XML documentation comments to explain its purpose, which is to serve as an in-memory test double for ISystemLeaseFactory with configurable lease duration for testing lease expiration scenarios.
| private sealed class InMemoryLeaseFactory : ISystemLeaseFactory | |
| { | |
| private readonly ConcurrentDictionary<string, InMemoryLease> leases = new(StringComparer.Ordinal); | |
| private readonly TimeSpan? overrideDuration; | |
| private long fencingToken; | |
| public InMemoryLeaseFactory(TimeSpan? overrideDuration = null) | |
| { | |
| this.overrideDuration = overrideDuration; | |
| } | |
| /// <summary> | |
| /// In-memory implementation of <see cref="ISystemLeaseFactory"/> used by tests. | |
| /// </summary> | |
| /// <remarks> | |
| /// This test double stores leases in memory and optionally forces a fixed lease | |
| /// duration via <paramref name="overrideDuration"/>. This allows tests to | |
| /// reliably simulate and verify lease expiration behavior without relying on | |
| /// external infrastructure or timing guarantees from a real lease store. | |
| /// </remarks> | |
| private sealed class InMemoryLeaseFactory : ISystemLeaseFactory | |
| { | |
| private readonly ConcurrentDictionary<string, InMemoryLease> leases = new(StringComparer.Ordinal); | |
| private readonly TimeSpan? overrideDuration; | |
| private long fencingToken; | |
| /// <summary> | |
| /// Initializes a new instance of the <see cref="InMemoryLeaseFactory"/> class. | |
| /// </summary> | |
| /// <param name="overrideDuration"> | |
| /// Optional lease duration to use for all acquired leases. When specified, | |
| /// this value overrides any duration passed to <see cref="AcquireAsync"/>. | |
| /// </param> | |
| public InMemoryLeaseFactory(TimeSpan? overrideDuration = null) | |
| { | |
| this.overrideDuration = overrideDuration; | |
| } | |
| /// <summary> | |
| /// Attempts to acquire an in-memory lease for the specified resource. | |
| /// </summary> | |
| /// <param name="resourceName">The name of the resource to acquire a lease for.</param> | |
| /// <param name="duration"> | |
| /// The requested lease duration. This is ignored if an <c>overrideDuration</c> | |
| /// was provided when constructing the factory. | |
| /// </param> | |
| /// <param name="contextJson">Optional JSON-encoded context associated with the lease.</param> | |
| /// <param name="ownerToken"> | |
| /// Optional owner token for the lease. When not provided, a new token is generated. | |
| /// </param> | |
| /// <param name="cancellationToken">A token to observe while waiting for the operation to complete.</param> | |
| /// <returns> | |
| /// A task that resolves to an <see cref="ISystemLease"/> instance if the lease | |
| /// was acquired, or <c>null</c> if another non-expired lease already exists | |
| /// for the same resource. | |
| /// </returns> |
|
|
||
| private sealed class StaticPlanner : IFanoutPlanner | ||
| { | ||
| private readonly IReadOnlyList<FanoutSlice> slices; | ||
|
|
||
| public StaticPlanner(IEnumerable<FanoutSlice> slices) | ||
| { | ||
| this.slices = slices.ToList(); | ||
| } | ||
|
|
There was a problem hiding this comment.
This test helper class lacks documentation. Consider adding XML documentation comments to explain its purpose, which is to provide a static list of FanoutSlice objects for testing scenarios where slice planning logic is not being tested.
| private sealed class StaticPlanner : IFanoutPlanner | |
| { | |
| private readonly IReadOnlyList<FanoutSlice> slices; | |
| public StaticPlanner(IEnumerable<FanoutSlice> slices) | |
| { | |
| this.slices = slices.ToList(); | |
| } | |
| /// <summary> | |
| /// Test helper implementation of <see cref="IFanoutPlanner"/> that always returns | |
| /// a preconfigured, static set of <see cref="FanoutSlice"/> instances. | |
| /// </summary> | |
| /// <remarks> | |
| /// This planner is used in scenarios where the fanout slice planning logic itself | |
| /// is not under test and a fixed set of slices is sufficient for exercising other | |
| /// parts of the fanout processing pipeline. | |
| /// </remarks> | |
| private sealed class StaticPlanner : IFanoutPlanner | |
| { | |
| private readonly IReadOnlyList<FanoutSlice> slices; | |
| /// <summary> | |
| /// Initializes a new instance of the <see cref="StaticPlanner"/> class with the | |
| /// provided collection of <see cref="FanoutSlice"/> objects. | |
| /// </summary> | |
| /// <param name="slices"> | |
| /// The slices that will be returned unchanged from <see cref="GetDueSlicesAsync"/>. | |
| /// </param> | |
| public StaticPlanner(IEnumerable<FanoutSlice> slices) | |
| { | |
| this.slices = slices.ToList(); | |
| } | |
| /// <inheritdoc /> |
| return Task.FromResult(slices); | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
This test helper class lacks documentation. Consider adding XML documentation comments to explain its purpose, which is to provide a sharded planner implementation for testing cursor-based slice filtering with a fixed set of shard keys.
| /// <summary> | |
| /// Test helper planner that enumerates a fixed set of shard keys for a given work key. | |
| /// This provides a sharded <see cref="BaseFanoutPlanner"/> implementation used to | |
| /// verify cursor-based slice filtering behavior with a stable, predetermined set of shards. | |
| /// </summary> |
| newLease.DisposeAsync().AsTask().Wait(); | ||
| return Task.FromResult<ISystemLease?>(null); | ||
| } | ||
|
|
||
| // If the resulting lease is expired, dispose the unused lease and loop to try again. | ||
| newLease.DisposeAsync().AsTask().Wait(); |
There was a problem hiding this comment.
Calling Wait() on an async dispose operation can lead to deadlocks, especially in test scenarios with synchronization contexts. Consider using ConfigureAwait(false).GetAwaiter().GetResult() instead, or restructure the code to avoid synchronous waits on async operations.
| newLease.DisposeAsync().AsTask().Wait(); | |
| return Task.FromResult<ISystemLease?>(null); | |
| } | |
| // If the resulting lease is expired, dispose the unused lease and loop to try again. | |
| newLease.DisposeAsync().AsTask().Wait(); | |
| newLease.DisposeAsync().AsTask().ConfigureAwait(false).GetAwaiter().GetResult(); | |
| return Task.FromResult<ISystemLease?>(null); | |
| } | |
| // If the resulting lease is expired, dispose the unused lease and loop to try again. | |
| newLease.DisposeAsync().AsTask().ConfigureAwait(false).GetAwaiter().GetResult(); |
| } | ||
|
|
||
| // If the resulting lease is expired, dispose the unused lease and loop to try again. | ||
| newLease.DisposeAsync().AsTask().Wait(); |
There was a problem hiding this comment.
Calling Wait() on an async dispose operation can lead to deadlocks, especially in test scenarios with synchronization contexts. Consider using ConfigureAwait(false).GetAwaiter().GetResult() instead, or restructure the code to avoid synchronous waits on async operations.
| while (true) | ||
| { | ||
| 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)) | ||
| { | ||
| // We successfully installed our lease. | ||
| return Task.FromResult<ISystemLease?>(newLease); | ||
| } | ||
|
|
||
| if (!resultingLease.IsExpired) | ||
| { | ||
| // Another thread holds a non-expired lease. | ||
| newLease.DisposeAsync().AsTask().Wait(); | ||
| return Task.FromResult<ISystemLease?>(null); | ||
| } | ||
|
|
||
| // If the resulting lease is expired, dispose the unused lease and loop to try again. | ||
| newLease.DisposeAsync().AsTask().Wait(); | ||
| } |
There was a problem hiding this comment.
The infinite while loop with concurrent AddOrUpdate calls has a potential race condition. Between checking if the resulting lease is expired (line 310) and looping to try again, another thread could acquire a new lease. This could cause the loop to spin indefinitely in highly contended scenarios. Consider adding a maximum retry count or a small delay between retries to prevent infinite spinning.
Summary
Testing
Codex Task