test(dynamodb): bound paginated conformance setup

src/Orleans.Clustering.TestKit/MembershipTableTestRunner.cs

@@ -12,20 +12,39 @@ public sealed class MembershipTableTestRunner
     private readonly int _seed;
     private readonly Action<string>? _output;
     private readonly int _concurrencyRowCount;
+    private readonly int _concurrencyRowPadding;
     private IMembershipTable A => _fixture.First;
     private IMembershipTable B => _fixture.Second;
     private IMembershipTable Other => _fixture.OtherCluster;
 
     /// <summary>Creates a runner over initialized independent handles. Row count controls workload, never guarantees.</summary>
     public MembershipTableTestRunner(MembershipTableTestFixture fixture, int seed = 0, Action<string>? output = null, int concurrencyRowCount = 128)
+        : this(fixture, seed, output, concurrencyRowCount, concurrencyRowPadding: 0)
+    {
+    }
+
+    /// <summary>Creates a runner over initialized independent handles with a configurable concurrent-read row payload.</summary>
+    /// <param name="fixture">The initialized fixture which owns the provider handles.</param>
+    /// <param name="seed">The deterministic test-data seed.</param>
+    /// <param name="output">Receives conformance diagnostics.</param>
+    /// <param name="concurrencyRowCount">The number of rows used by concurrent-read scenarios.</param>
+    /// <param name="concurrencyRowPadding">The number of characters appended to each concurrent-read row's host name.</param>
+    public MembershipTableTestRunner(
+        MembershipTableTestFixture fixture,
+        int seed,
+        Action<string>? 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;
     }
 
     /// <summary>G01: insertion and its version are one exact +1 commit, preserving the sentinel.</summary>
@@ -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;

src/api/Orleans.Clustering.TestKit/Orleans.Clustering.TestKit.cs

@@ -77,6 +77,8 @@ public MembershipTableTestHandle(IMembershipTable table, System.Func<System.Thre
 
     public sealed partial class MembershipTableTestRunner
     {
+        public MembershipTableTestRunner(MembershipTableTestFixture fixture, int seed, System.Action<string>? output, int concurrencyRowCount, int concurrencyRowPadding) { }
+
         public MembershipTableTestRunner(MembershipTableTestFixture fixture, int seed = 0, System.Action<string>? output = null, int concurrencyRowCount = 128) { }
 
         public System.Threading.Tasks.Task CleanupDefunctSiloEntries_RemovesOnlyStrictlyOldDeadRows(System.Threading.CancellationToken cancellationToken = default) { throw null; }

test/Extensions/Orleans.AWS.Tests/MembershipTests/DynamoDBMembershipTableTest.cs

@@ -26,6 +26,9 @@ namespace AWSUtils.Tests.MembershipTests
     [TestArea("Membership")]
     public class DynamoDBMembershipTableTest : MembershipTableTestsBase, IClassFixture<DynamoDBStorageTestsFixture>
     {
+        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()
         {

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()
         {

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]

test/Orleans.Clustering.TestKit.Tests/MembershipTableTestFixtureTests.cs

@@ -540,6 +540,8 @@ public async Task PublicArguments_AreValidatedBeforeConstructingProviders()
         Assert.Throws<ArgumentException>(() => new MembershipTableTestRunner(fixture));
         await fixture.InitializeAsync(TestContext.Current.CancellationToken);
         Assert.Throws<ArgumentOutOfRangeException>(() => new MembershipTableTestRunner(fixture, concurrencyRowCount: 2));
+        Assert.Throws<ArgumentOutOfRangeException>(() => new MembershipTableTestRunner(fixture, 0, null, 3, -1));
+        Assert.Throws<ArgumentOutOfRangeException>(() => new MembershipTableTestRunner(fixture, 0, null, 3, 256 * 1024 + 1));
         Assert.Equal(3, backend.CreatedHandles);
         await fixture.DisposeAsync();
         await Assert.ThrowsAsync<ObjectDisposedException>(() => fixture.InitializeAsync(TestContext.Current.CancellationToken).AsTask());

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<MembershipTableTestRunner, CancellationToken, T
                     fixture,
                     seed: 17,
                     output: WriteConformanceOutput,
-                    concurrencyRowCount: ConformanceConcurrencyRowCount),
+                    concurrencyRowCount: ConformanceConcurrencyRowCount,
+                    concurrencyRowPadding: ConformanceConcurrencyRowPadding),
                 cancellationToken),
             TestContext.Current.CancellationToken);