diff --git a/src/Orleans.Clustering.TestKit/MembershipTableTestRunner.cs b/src/Orleans.Clustering.TestKit/MembershipTableTestRunner.cs index b157acc95a5..3ca9643c00a 100644 --- a/src/Orleans.Clustering.TestKit/MembershipTableTestRunner.cs +++ b/src/Orleans.Clustering.TestKit/MembershipTableTestRunner.cs @@ -12,20 +12,39 @@ public sealed class MembershipTableTestRunner private readonly int _seed; private readonly Action? _output; private readonly int _concurrencyRowCount; + private readonly int _concurrencyRowPadding; private IMembershipTable A => _fixture.First; private IMembershipTable B => _fixture.Second; private IMembershipTable Other => _fixture.OtherCluster; /// Creates a runner over initialized independent handles. Row count controls workload, never guarantees. public MembershipTableTestRunner(MembershipTableTestFixture fixture, int seed = 0, Action? output = null, int concurrencyRowCount = 128) + : this(fixture, seed, output, concurrencyRowCount, concurrencyRowPadding: 0) + { + } + + /// Creates a runner over initialized independent handles with a configurable concurrent-read row payload. + /// The initialized fixture which owns the provider handles. + /// The deterministic test-data seed. + /// Receives conformance diagnostics. + /// The number of rows used by concurrent-read scenarios. + /// The number of characters appended to each concurrent-read row's host name. + public MembershipTableTestRunner( + MembershipTableTestFixture fixture, + int seed, + Action? output, + int concurrencyRowCount, + int concurrencyRowPadding) { _fixture = fixture ?? throw new ArgumentNullException(nameof(fixture)); if (fixture.First is null || fixture.Second is null || fixture.OtherCluster is null) throw new ArgumentException("Initialize the fixture before constructing its runner.", nameof(fixture)); if (concurrencyRowCount is < 3 or > 10000) throw new ArgumentOutOfRangeException(nameof(concurrencyRowCount)); + if (concurrencyRowPadding is < 0 or > 256 * 1024) throw new ArgumentOutOfRangeException(nameof(concurrencyRowPadding)); _seed = seed; _output = output; _concurrencyRowCount = concurrencyRowCount; + _concurrencyRowPadding = concurrencyRowPadding; } /// G01: insertion and its version are one exact +1 commit, preserving the sentinel. @@ -687,9 +706,14 @@ internal async Task SeedConcurrentRows(CancellationToken ct) var current = ClusteringMembershipSnapshot.Capture(await A.ReadAllAsync(ct)); Check(current.Rows.Count == 0, "concurrent-read setup requires an empty cluster"); var expectedRows = current.Rows.ToBuilder(); + // Each insert needs the new opaque table ETag, so setup performs a growing full read after every row. for (var i = 1; i <= _concurrencyRowCount; i++) { var input = Entry(i); + if (_concurrencyRowPadding > 0) + { + input.HostName += new string('x', _concurrencyRowPadding); + } var expectedEntry = MembershipEntrySnapshot.Capture(input); var writer = i % 2 == 0 ? B : A; var reader = i % 2 == 0 ? A : B; diff --git a/src/api/Orleans.Clustering.TestKit/Orleans.Clustering.TestKit.cs b/src/api/Orleans.Clustering.TestKit/Orleans.Clustering.TestKit.cs index 460f8d2fa67..f8a16539337 100644 --- a/src/api/Orleans.Clustering.TestKit/Orleans.Clustering.TestKit.cs +++ b/src/api/Orleans.Clustering.TestKit/Orleans.Clustering.TestKit.cs @@ -77,6 +77,8 @@ public MembershipTableTestHandle(IMembershipTable table, System.Func? output, int concurrencyRowCount, int concurrencyRowPadding) { } + public MembershipTableTestRunner(MembershipTableTestFixture fixture, int seed = 0, System.Action? output = null, int concurrencyRowCount = 128) { } public System.Threading.Tasks.Task CleanupDefunctSiloEntries_RemovesOnlyStrictlyOldDeadRows(System.Threading.CancellationToken cancellationToken = default) { throw null; } diff --git a/test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableTest.cs b/test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableTest.cs index 2e871489bcf..07b7dffceba 100644 --- a/test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableTest.cs +++ b/test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableTest.cs @@ -26,6 +26,9 @@ namespace AWSUtils.Tests.MembershipTests [TestArea("Membership")] public class DynamoDBMembershipTableTest : MembershipTableTestsBase, IClassFixture { + internal const int ConcurrentReadRowCount = 48; + internal const int ConcurrentReadRowPadding = 40 * 1024; + public DynamoDBMembershipTableTest(ConnectionStringFixture fixture, TestEnvironmentFixture environment) : base(fixture, environment, CreateFilters()) { } @@ -51,8 +54,10 @@ protected override IMembershipTable CreateMembershipTable(ILogger logger, IOptio return new TestOwnedDynamoDBMembershipTable(this.loggerFactory, Options.Create(options), clusterOptions); } - // Persisted fields and suspect votes exceed DynamoDB's 1 MiB query page at this count. - protected override int ConformanceConcurrencyRowCount => 4096; + // Large rows cross DynamoDB's 1 MiB query boundary while keeping full-snapshot setup bounded. + // ReadAllRetriesWhenCanonicalMutationCrossesPages provides deterministic inter-page retry coverage. + protected override int ConformanceConcurrencyRowCount => ConcurrentReadRowCount; + protected override int ConformanceConcurrencyRowPadding => ConcurrentReadRowPadding; protected override MembershipTableTestFixture CreateConformanceFixture() { diff --git a/test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableUnitTests.cs b/test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableUnitTests.cs index 3db90e87039..00e8cb0f01f 100644 --- a/test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableUnitTests.cs +++ b/test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableUnitTests.cs @@ -1069,6 +1069,16 @@ public async Task ReadAllRetriesWhenCanonicalMutationCrossesPages(bool versionFi Assert.Equal("host-8", Assert.Single(result.Members).Item1.HostName); } + [Fact] + public void ConcurrentReadConformanceWorkloadCrossesQueryPageAfterCleanup() + { + const int maximumQueryPageBytes = 1024 * 1024; + const int cleanedRows = 2; + Assert.True( + (DynamoDBMembershipTableTest.ConcurrentReadRowCount - cleanedRows) * DynamoDBMembershipTableTest.ConcurrentReadRowPadding + > maximumQueryPageBytes); + } + [Fact] public async Task ReadAllReturnsCompleteViewWhenMutationFollowsQuery() { diff --git a/test/Orleans.Clustering.TestKit.Tests/MembershipTableConformanceTests.cs b/test/Orleans.Clustering.TestKit.Tests/MembershipTableConformanceTests.cs index 7e23393cef3..9edfb3ecc5d 100644 --- a/test/Orleans.Clustering.TestKit.Tests/MembershipTableConformanceTests.cs +++ b/test/Orleans.Clustering.TestKit.Tests/MembershipTableConformanceTests.cs @@ -95,6 +95,26 @@ await backend.Fixture().RunAsync(async (fixture, ct) => Assert.Equal($"host-{i}", partition.Rows[MembershipTableTestData.CreateEntry(i).SiloAddress].Item1.HostName); }, TestContext.Current.CancellationToken); } + + [Fact] + public async Task ConcurrentReadSetup_AppliesConfiguredRowPadding() + { + const int rows = 5; + const int rowPadding = 1024; + var backend = new IdealizedMembershipBackend(); + await backend.Fixture().RunAsync(async (fixture, ct) => + { + await new MembershipTableTestRunner(fixture, seed: 0, output: null, concurrencyRowCount: rows, concurrencyRowPadding: rowPadding) + .SeedConcurrentRows(ct); + var partition = backend.Partitions[fixture.ClusterId]; + for (var i = 1; i <= rows; i++) + { + Assert.Equal( + $"host-{i}{new string('x', rowPadding)}", + partition.Rows[MembershipTableTestData.CreateEntry(i).SiloAddress].Item1.HostName); + } + }, TestContext.Current.CancellationToken); + } [Fact] public Task InitializeMembershipTable_RepeatedWithData_PreservesCommittedState() => Run((r, ct) => r.InitializeMembershipTable_RepeatedWithData_PreservesCommittedState(ct)); [Fact] diff --git a/test/Orleans.Clustering.TestKit.Tests/MembershipTableTestFixtureTests.cs b/test/Orleans.Clustering.TestKit.Tests/MembershipTableTestFixtureTests.cs index 17bab8f7728..ce0e194ddeb 100644 --- a/test/Orleans.Clustering.TestKit.Tests/MembershipTableTestFixtureTests.cs +++ b/test/Orleans.Clustering.TestKit.Tests/MembershipTableTestFixtureTests.cs @@ -540,6 +540,8 @@ public async Task PublicArguments_AreValidatedBeforeConstructingProviders() Assert.Throws(() => new MembershipTableTestRunner(fixture)); await fixture.InitializeAsync(TestContext.Current.CancellationToken); Assert.Throws(() => new MembershipTableTestRunner(fixture, concurrencyRowCount: 2)); + Assert.Throws(() => new MembershipTableTestRunner(fixture, 0, null, 3, -1)); + Assert.Throws(() => new MembershipTableTestRunner(fixture, 0, null, 3, 256 * 1024 + 1)); Assert.Equal(3, backend.CreatedHandles); await fixture.DisposeAsync(); await Assert.ThrowsAsync(() => fixture.InitializeAsync(TestContext.Current.CancellationToken).AsTask()); diff --git a/test/Orleans.Runtime.Internal.Tests/MembershipTests/MembershipTableConformanceTestsBase.cs b/test/Orleans.Runtime.Internal.Tests/MembershipTests/MembershipTableConformanceTestsBase.cs index 6ea972970b1..58a934a9113 100644 --- a/test/Orleans.Runtime.Internal.Tests/MembershipTests/MembershipTableConformanceTestsBase.cs +++ b/test/Orleans.Runtime.Internal.Tests/MembershipTests/MembershipTableConformanceTestsBase.cs @@ -10,6 +10,8 @@ public abstract class MembershipTableConformanceTestsBase protected abstract MembershipTableTestFixture CreateConformanceFixture(); protected virtual int ConformanceConcurrencyRowCount => 128; + // Padding is appended to HostName and must fit the provider's persisted string limit. + protected virtual int ConformanceConcurrencyRowPadding => 0; protected virtual void WriteConformanceOutput(string message) => TestContext.Current.TestOutputHelper?.WriteLine(message); @@ -33,7 +35,8 @@ private Task RunConformance(Func