Skip to content

Absorb the sharded scrub template in TenantPartitionedEventsTests/Sharded #5103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jeremydmiller merged 1 commit into from
Aug 1, 2026
Merged

Absorb the sharded scrub template in TenantPartitionedEventsTests/Sharded #5103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 18 additions & 37 deletions src/TenantPartitionedEventsTests/Sharded/Bug_4679_sharded_catch_up_23505.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,34 +55,15 @@ namespace TenantPartitionedEventsTests.Sharded;
/// </para>
/// </summary>
[Collection("sharded-tenant-partitioned")]
public partial class Bug_4679_sharded_catch_up_23505: IAsyncLifetime
public partial class Bug_4679_sharded_catch_up_23505: ShardedPartitionedContext
{
private readonly ShardedPartitionedFixture _fixture;
private readonly ITestOutputHelper _output;

public Bug_4679_sharded_catch_up_23505(ShardedPartitionedFixture fixture, ITestOutputHelper output)
public Bug_4679_sharded_catch_up_23505(ShardedPartitionedFixture fixture, ITestOutputHelper output): base(fixture)
{
_fixture = fixture;
_output = output;
}

public async ValueTask InitializeAsync()
{
await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString);
await conn.OpenAsync();
try { await conn.DropSchemaAsync("sharded"); } catch { }

foreach (var connStr in _fixture.ConnectionStrings.Values)
{
await using var tenantConn = new NpgsqlConnection(connStr);
await tenantConn.OpenAsync();
try { await tenantConn.DropSchemaAsync("tenants"); } catch { }
await ShardedPartitionedFixture.CleanMartenObjectsInPublicSchema(tenantConn);
}
}

public ValueTask DisposeAsync() => default;

public class Bug4679Trip
{
public Guid Id { get; set; }
Expand Down Expand Up @@ -130,7 +111,7 @@ private void configure(StoreOptions opts)
x.SchemaName = "sharded";
x.PartitionSchemaName = "tenants";

foreach (var (dbName, connStr) in _fixture.ConnectionStrings)
foreach (var (dbName, connStr) in Fixture.ConnectionStrings)
{
x.AddDatabase(dbName, connStr);
}
Expand Down Expand Up @@ -159,14 +140,14 @@ public async Task force_catch_up_with_multiple_tenants_on_one_shard()
var tenants = new[] { "tA", "tB", "tC", "tD", "tE" };
var shardAssignment = new Dictionary<string, string>
{
["tA"] = _fixture.DbNames[0],
["tB"] = _fixture.DbNames[0],
["tC"] = _fixture.DbNames[0],
["tD"] = _fixture.DbNames[1],
["tE"] = _fixture.DbNames[2],
["tA"] = Fixture.DbNames[0],
["tB"] = Fixture.DbNames[0],
["tC"] = Fixture.DbNames[0],
["tD"] = Fixture.DbNames[1],
["tE"] = Fixture.DbNames[2],
};

await using var store = (DocumentStore)DocumentStore.For(configure);
var store = TrackStore((DocumentStore)DocumentStore.For(configure));
foreach (var tenant in tenants)
{
await store.Advanced.AddTenantToShardAsync(tenant, shardAssignment[tenant], CancellationToken.None);
Expand Down Expand Up @@ -210,7 +191,7 @@ public async Task force_catch_up_with_multiple_tenants_on_one_shard()
// TENANT-SCOPED (…:All:{tenant}); there is NO bare store-global …:All row (which is what
// the user reports colliding). This is the guard that would flip red if the per-tenant
// catch-up ever started writing store-global progression names.
var names = await progressionNamesAsync(_fixture.ConnectionStrings[shardAssignment["tA"]]);
var names = await progressionNamesAsync(Fixture.ConnectionStrings[shardAssignment["tA"]]);
names.ShouldContain("Bug4679ShardedTrip:All:tA");
names.ShouldContain("Bug4679ShardedTrip:All:tB");
names.ShouldNotContain("Bug4679ShardedTrip:All");
Expand All @@ -225,7 +206,7 @@ private void configureComposite(StoreOptions opts)
x.SchemaName = "sharded";
x.PartitionSchemaName = "tenants";

foreach (var (dbName, connStr) in _fixture.ConnectionStrings)
foreach (var (dbName, connStr) in Fixture.ConnectionStrings)
{
x.AddDatabase(dbName, connStr);
}
Expand Down Expand Up @@ -255,14 +236,14 @@ public async Task composite_force_catch_up_with_multiple_tenants_on_one_shard()
var tenants = new[] { "tA", "tB", "tC", "tD", "tE" };
var shardAssignment = new Dictionary<string, string>
{
["tA"] = _fixture.DbNames[0],
["tB"] = _fixture.DbNames[0],
["tC"] = _fixture.DbNames[0],
["tD"] = _fixture.DbNames[1],
["tE"] = _fixture.DbNames[2],
["tA"] = Fixture.DbNames[0],
["tB"] = Fixture.DbNames[0],
["tC"] = Fixture.DbNames[0],
["tD"] = Fixture.DbNames[1],
["tE"] = Fixture.DbNames[2],
};

await using var store = (DocumentStore)DocumentStore.For(configureComposite);
var store = TrackStore((DocumentStore)DocumentStore.For(configureComposite));
foreach (var tenant in tenants)
{
await store.Advanced.AddTenantToShardAsync(tenant, shardAssignment[tenant], CancellationToken.None);
Expand Down Expand Up @@ -294,7 +275,7 @@ public async Task composite_force_catch_up_with_multiple_tenants_on_one_shard()
_output.WriteLine(e.ToString());
}

var names = await progressionNamesAsync(_fixture.ConnectionStrings[shardAssignment["tA"]]);
var names = await progressionNamesAsync(Fixture.ConnectionStrings[shardAssignment["tA"]]);
_output.WriteLine("=== mt_event_progression rows on shard_a ===");
foreach (var n in names) _output.WriteLine(n);

Expand Down
38 changes: 10 additions & 28 deletions src/TenantPartitionedEventsTests/Sharded/Bug_4706_sharded_partitioned_doc_rebuild.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,41 +30,23 @@ public class Bug4706Doc
/// is fine. A single-DB managed-ByList store is idempotent; this exercises the sharded path.
/// </summary>
[Collection("sharded-tenant-partitioned")]
public class Bug_4706_sharded_partitioned_doc_rebuild: IAsyncLifetime
public class Bug_4706_sharded_partitioned_doc_rebuild: ShardedPartitionedContext
{
private readonly ShardedPartitionedFixture _fixture;
private readonly ITestOutputHelper _output;

public Bug_4706_sharded_partitioned_doc_rebuild(ShardedPartitionedFixture fixture, ITestOutputHelper output)
public Bug_4706_sharded_partitioned_doc_rebuild(ShardedPartitionedFixture fixture, ITestOutputHelper output): base(fixture)
{
_fixture = fixture;
_output = output;
}

public async ValueTask InitializeAsync()
{
await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString);
await conn.OpenAsync();
try { await conn.DropSchemaAsync("sharded"); } catch { }
foreach (var connStr in _fixture.ConnectionStrings.Values)
{
await using var tenantConn = new NpgsqlConnection(connStr);
await tenantConn.OpenAsync();
try { await tenantConn.DropSchemaAsync("tenants"); } catch { }
await ShardedPartitionedFixture.CleanMartenObjectsInPublicSchema(tenantConn);
}
}

public ValueTask DisposeAsync() => default;

private DocumentStore BuildStore() => (DocumentStore)DocumentStore.For(opts =>
private DocumentStore BuildStore() => TrackStore((DocumentStore)DocumentStore.For(opts =>
{
opts.MultiTenantedWithShardedDatabases(x =>
{
x.ConnectionString = ConnectionSource.ConnectionString;
x.SchemaName = "sharded";
x.PartitionSchemaName = "tenants";
foreach (var (dbName, connStr) in _fixture.ConnectionStrings)
foreach (var (dbName, connStr) in Fixture.ConnectionStrings)
{
x.AddDatabase(dbName, connStr);
}
Expand All @@ -78,24 +60,24 @@ private DocumentStore BuildStore() => (DocumentStore)DocumentStore.For(opts =>
.MultiTenantedWithPartitioning(x => x.ByList())
.Index(x => x.Name)
.StartIndexesByTenantId();
});
}));

[Fact]
public async Task reapply_over_existing_tenant_data_is_idempotent()
{
var assignment = new Dictionary<string, string>
{
["tA"] = _fixture.DbNames[0],
["tB"] = _fixture.DbNames[1],
["tC"] = _fixture.DbNames[2],
["tA"] = Fixture.DbNames[0],
["tB"] = Fixture.DbNames[1],
["tC"] = Fixture.DbNames[2],
};

// First deploy: production eager-apply shape — create the parent partitioned
// schema on every shard up front (per database, sequentially), THEN provision
// tenants (which adds each tenant's partition to its shard via the additive
// path), THEN write data.
await using (var store = BuildStore())
{
var store = BuildStore();
var databases = await store.Options.Tenancy.BuildDatabases();
foreach (var db in databases.OfType<IMartenDatabase>())
{
Expand All @@ -118,8 +100,8 @@ public async Task reapply_over_existing_tenant_data_is_idempotent()
// Second deploy (nothing changed): re-apply across all shards via the store-wide
// path (Parallel.ForEachAsync over databases — the #4706 trigger). Must be
// idempotent: no destructive rebuild of the per-tenant partitioned tables, no 23514.
await using (var store = BuildStore())
{
var store = BuildStore();
var ex = await Record.ExceptionAsync(() =>
store.Storage.ApplyAllConfiguredChangesToDatabaseAsync());

Expand Down
42 changes: 12 additions & 30 deletions src/TenantPartitionedEventsTests/Sharded/Bug_4713_sharded_reapply_same_store.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,41 +43,23 @@ public class Bug4713Doc
/// </para>
/// </summary>
[Collection("sharded-tenant-partitioned")]
public class Bug_4713_sharded_reapply_same_store: IAsyncLifetime
public class Bug_4713_sharded_reapply_same_store: ShardedPartitionedContext
{
private readonly ShardedPartitionedFixture _fixture;
private readonly ITestOutputHelper _output;

public Bug_4713_sharded_reapply_same_store(ShardedPartitionedFixture fixture, ITestOutputHelper output)
public Bug_4713_sharded_reapply_same_store(ShardedPartitionedFixture fixture, ITestOutputHelper output): base(fixture)
{
_fixture = fixture;
_output = output;
}

public async ValueTask InitializeAsync()
{
await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString);
await conn.OpenAsync();
try { await conn.DropSchemaAsync("sharded"); } catch { }
foreach (var connStr in _fixture.ConnectionStrings.Values)
{
await using var tenantConn = new NpgsqlConnection(connStr);
await tenantConn.OpenAsync();
try { await tenantConn.DropSchemaAsync("tenants"); } catch { }
await ShardedPartitionedFixture.CleanMartenObjectsInPublicSchema(tenantConn);
}
}

public ValueTask DisposeAsync() => default;

private DocumentStore BuildStore() => (DocumentStore)DocumentStore.For(opts =>
private DocumentStore BuildStore() => TrackStore((DocumentStore)DocumentStore.For(opts =>
{
opts.MultiTenantedWithShardedDatabases(x =>
{
x.ConnectionString = ConnectionSource.ConnectionString;
x.SchemaName = "sharded";
x.PartitionSchemaName = "tenants";
foreach (var (dbName, connStr) in _fixture.ConnectionStrings)
foreach (var (dbName, connStr) in Fixture.ConnectionStrings)
{
x.AddDatabase(dbName, connStr);
}
Expand All @@ -91,20 +73,20 @@ private DocumentStore BuildStore() => (DocumentStore)DocumentStore.For(opts =>
.MultiTenantedWithPartitioning(x => x.ByList())
.Index(x => x.Name)
.StartIndexesByTenantId();
});
}));

[Fact]
public async Task reapply_on_same_store_after_provisioning_is_idempotent()
{
var assignment = new Dictionary<string, string>
{
["tA"] = _fixture.DbNames[0],
["tB"] = _fixture.DbNames[1],
["tC"] = _fixture.DbNames[2],
["tA"] = Fixture.DbNames[0],
["tB"] = Fixture.DbNames[1],
["tC"] = Fixture.DbNames[2],
};

// ONE store instance for the whole test — this is the distinguishing factor from #4706.
await using var store = BuildStore();
var store = BuildStore();

var databases = await store.Options.Tenancy.BuildDatabases();
foreach (var db in databases.OfType<IMartenDatabase>())
Expand All @@ -128,11 +110,11 @@ public async Task reapply_on_same_store_after_provisioning_is_idempotent()

// Capture the partition layout each shard has AFTER provisioning. #4713 is specifically about
// re-apply IDEMPOTENCY, so this is the baseline a no-op re-apply must preserve.
var shards = _fixture.ConnectionStrings.Keys.ToList();
var shards = Fixture.ConnectionStrings.Keys.ToList();
var before = new Dictionary<string, List<string>>();
foreach (var shard in shards)
{
before[shard] = await partitionsOf(_fixture.ConnectionStrings[shard], "mt_doc_bug4713doc");
before[shard] = await partitionsOf(Fixture.ConnectionStrings[shard], "mt_doc_bug4713doc");
_output.WriteLine($"[before][{shard}] {string.Join(", ", before[shard])}");
}

Expand All @@ -156,7 +138,7 @@ public async Task reapply_on_same_store_after_provisioning_is_idempotent()
// #4713 fix restores (flag stays set → ListPartitioning.CreateDelta → None on every shard).
foreach (var shard in shards)
{
var after = await partitionsOf(_fixture.ConnectionStrings[shard], "mt_doc_bug4713doc");
var after = await partitionsOf(Fixture.ConnectionStrings[shard], "mt_doc_bug4713doc");
_output.WriteLine($"[after][{shard}] {string.Join(", ", after)}");
after.ShouldBe(before[shard],
$"re-applying the schema must not change shard {shard}'s per-tenant partitions (#4713)");
Expand Down
33 changes: 6 additions & 27 deletions src/TenantPartitionedEventsTests/Sharded/Bug_4751_composite_catchup_under_sharded.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,48 +63,27 @@ public partial class CmpTripCountProjection: SingleStreamProjection<CmpTripCount
/// </para>
/// </summary>
[Collection("sharded-tenant-partitioned")]
public class Bug_4751_composite_catchup_under_sharded: IAsyncLifetime
public class Bug_4751_composite_catchup_under_sharded: ShardedPartitionedContext
{
private readonly ShardedPartitionedFixture _fixture;
private readonly ITestOutputHelper _output;
private DocumentStore _store = null!;

public Bug_4751_composite_catchup_under_sharded(ShardedPartitionedFixture fixture, ITestOutputHelper output)
public Bug_4751_composite_catchup_under_sharded(ShardedPartitionedFixture fixture, ITestOutputHelper output): base(fixture)
{
_fixture = fixture;
_output = output;
}

public async ValueTask InitializeAsync()
{
await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString);
await conn.OpenAsync();
try { await conn.DropSchemaAsync("sharded"); } catch { }
foreach (var connStr in _fixture.ConnectionStrings.Values)
{
await using var tenantConn = new NpgsqlConnection(connStr);
await tenantConn.OpenAsync();
try { await tenantConn.DropSchemaAsync("tenants"); } catch { }
await ShardedPartitionedFixture.CleanMartenObjectsInPublicSchema(tenantConn);
}
}

public async ValueTask DisposeAsync()
{
if (_store != null!) await _store.DisposeAsync();
}

[Fact]
public async Task composite_reaches_non_stale_via_normal_daemon_catchup()
{
_store = (DocumentStore)DocumentStore.For(opts =>
_store = TrackStore((DocumentStore)DocumentStore.For(opts =>
{
opts.MultiTenantedWithShardedDatabases(x =>
{
x.ConnectionString = ConnectionSource.ConnectionString;
x.SchemaName = "sharded";
x.PartitionSchemaName = "tenants";
foreach (var (dbName, connStr) in _fixture.ConnectionStrings)
foreach (var (dbName, connStr) in Fixture.ConnectionStrings)
{
x.AddDatabase(dbName, connStr);
}
Expand All @@ -124,9 +103,9 @@ public async Task composite_reaches_non_stale_via_normal_daemon_catchup()

opts.Schema.For<CmpTrip>().DocumentAlias("cmp_trip");
opts.Schema.For<CmpTripCount>().DocumentAlias("cmp_trip_count");
});
}));

var shard = _fixture.DbNames[0];
var shard = Fixture.DbNames[0];
await _store.Advanced.AddTenantToShardAsync("tenant_a", shard, CancellationToken.None);

const int streams = 10;
Expand Down
Loading
Loading