diff --git a/src/TenantPartitionedEventsTests/Admin/admin_extras_under_partitioning.cs b/src/TenantPartitionedEventsTests/Admin/admin_extras_under_partitioning.cs index bc576a7142..31de334baa 100644 --- a/src/TenantPartitionedEventsTests/Admin/admin_extras_under_partitioning.cs +++ b/src/TenantPartitionedEventsTests/Admin/admin_extras_under_partitioning.cs @@ -11,6 +11,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -31,36 +32,12 @@ namespace TenantPartitionedEventsTests.Admin; /// sequences after registering N tenants; re-apply is idempotent. /// /// -public class admin_extras_under_partitioning : IAsyncLifetime +public class admin_extras_under_partitioning : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_xtra"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_xtra_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); - - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; } [Fact] @@ -81,15 +58,15 @@ public async Task AddMartenManagedTenantsAsync_with_Guids_uses_hyphenfree_N_form var tenantGuid = Guid.NewGuid(); var expectedSuffix = tenantGuid.ToString("N"); // 32 hex chars, no hyphens - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenantGuid); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenantGuid); - var sequenceExists = await SequenceExistsAsync(_schema, $"mt_events_sequence_{expectedSuffix}"); + var sequenceExists = await SequenceExistsAsync(Schema, $"mt_events_sequence_{expectedSuffix}"); sequenceExists.ShouldBeTrue( $"per-tenant sequence must be named mt_events_sequence_{expectedSuffix} (N-format) — " + "this pins the hyphen-free format choice from #4567"); // Sanity: the hyphenated "D" form is NOT used. - var hyphenatedShouldNotExist = await SequenceExistsAsync(_schema, + var hyphenatedShouldNotExist = await SequenceExistsAsync(Schema, $"mt_events_sequence_{tenantGuid.ToString("D")}"); hyphenatedShouldNotExist.ShouldBeFalse( "the hyphenated D format must NOT be used — N is the canonical choice"); @@ -103,14 +80,14 @@ public async Task AssertDatabaseMatchesConfigurationAsync_reports_no_drift_after // FK (#4606) and the per-tenant sequence count must NOT register as // drift items — those are intentional shape choices, not missing // schema objects. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); // After EnsureStorageExistsAsync (in InitializeAsync) + tenant // registration, the on-disk schema and the configuration should be // byte-identical. AssertDatabaseMatchesConfigurationAsync throws when // they diverge; Should.NotThrowAsync pins the no-drift contract. await Should.NotThrowAsync(async () => - await _store.Storage.Database.AssertDatabaseMatchesConfigurationAsync()); + await Store.Storage.Database.AssertDatabaseMatchesConfigurationAsync()); } [Fact] @@ -121,15 +98,15 @@ public async Task PerTenantEventSequences_exactly_one_sequence_per_registered_te // store-global mt_events_sequence). Re-applying the schema (via // EnsureStorageExistsAsync again) is idempotent — no new sequences, // no duplicates. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "one", "two", "three"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "one", "two", "three"); - var sequencesAfterRegistration = await CountTenantSequencesAsync(_schema); + var sequencesAfterRegistration = await CountTenantSequencesAsync(Schema); sequencesAfterRegistration.ShouldBe(3L); // Idempotency: re-applying changes shouldn't change the count. - await _store.Storage.Database.ApplyAllConfiguredChangesToDatabaseAsync(ct: TestContext.Current.CancellationToken); + await Store.Storage.Database.ApplyAllConfiguredChangesToDatabaseAsync(ct: TestContext.Current.CancellationToken); - var sequencesAfterReapply = await CountTenantSequencesAsync(_schema); + var sequencesAfterReapply = await CountTenantSequencesAsync(Schema); sequencesAfterReapply.ShouldBe(3L, "PerTenantEventSequences emits CREATE SEQUENCE IF NOT EXISTS — re-apply must be idempotent, no duplicates"); } diff --git a/src/TenantPartitionedEventsTests/Admin/delete_all_tenant_data_orphan_sequence_pin.cs b/src/TenantPartitionedEventsTests/Admin/delete_all_tenant_data_orphan_sequence_pin.cs index 3a30784f3f..eddd38c48b 100644 --- a/src/TenantPartitionedEventsTests/Admin/delete_all_tenant_data_orphan_sequence_pin.cs +++ b/src/TenantPartitionedEventsTests/Admin/delete_all_tenant_data_orphan_sequence_pin.cs @@ -14,6 +14,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -29,71 +30,46 @@ namespace TenantPartitionedEventsTests.Admin; /// match) is the dropped tenant. /// Store-global progression rows are intentionally left alone. /// -public class delete_all_tenant_data_orphan_sequence_pin : IAsyncLifetime +public class delete_all_tenant_data_orphan_sequence_pin : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_del"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_del_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); - - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - // #4683 progression test wants an async projection so the rebuild populates - // per-tenant mt_event_progression rows we can then assert on. - opts.Projections.Add(ProjectionLifecycle.Async); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Events.AddEventType(); + // #4683 progression test wants an async projection so the rebuild populates + // per-tenant mt_event_progression rows we can then assert on. + opts.Projections.Add(ProjectionLifecycle.Async); } [Fact] public async Task DeleteAllTenantDataAsync_drops_partitions_and_the_per_tenant_sequence() { var tenant = "delpin"; - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); // Seed some events so the per-tenant sequence advances. var streamId = Guid.NewGuid(); - await using (var session = _store.LightweightSession(tenant)) + await using (var session = Store.LightweightSession(tenant)) { session.Events.StartStream(streamId, new DelEvent("a"), new DelEvent("b"), new DelEvent("c")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - var seqValueBefore = await ReadSequenceLastValueAsync(_schema, $"mt_events_sequence_{tenant}"); + var seqValueBefore = await ReadSequenceLastValueAsync(Schema, $"mt_events_sequence_{tenant}"); seqValueBefore.ShouldBeGreaterThanOrEqualTo(3L, "the sequence advanced past the 3 appended events"); // Act: delete all data for this tenant. - await _store.Advanced.DeleteAllTenantDataAsync(tenant, CancellationToken.None); + await Store.Advanced.DeleteAllTenantDataAsync(tenant, CancellationToken.None); // The partition tables are gone — pinning the cleaner's positive effect. - var partitionExists = await TableExistsAsync(_schema, $"mt_events_{tenant}"); + var partitionExists = await TableExistsAsync(Schema, $"mt_events_{tenant}"); partitionExists.ShouldBeFalse( "DeleteAllTenantDataAsync drops the tenant's mt_events partition table"); // #4683: per-tenant sequence is now dropped (was the orphan-leak pin). - var seqStillExists = await SequenceExistsAsync(_schema, $"mt_events_sequence_{tenant}"); + var seqStillExists = await SequenceExistsAsync(Schema, $"mt_events_sequence_{tenant}"); seqStillExists.ShouldBeFalse( "DeleteAllTenantDataAsync now drops the per-tenant mt_events_sequence_ via " + "PerTenantPartitionedCleanup (#4683). Was previously pinned as the orphan leak."); @@ -107,26 +83,26 @@ public async Task RemoveMartenManagedTenantsAsync_drops_partitions_and_the_per_t // explicit "I no longer need this tenant" route), so this confirms PerTenantPartitionedCleanup // is wired in on *both* paths -- not just the cleaner's. var tenant = "rempin"; - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); var streamId = Guid.NewGuid(); - await using (var session = _store.LightweightSession(tenant)) + await using (var session = Store.LightweightSession(tenant)) { session.Events.StartStream(streamId, new DelEvent("x"), new DelEvent("y")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - var seqValueBefore = await ReadSequenceLastValueAsync(_schema, $"mt_events_sequence_{tenant}"); + var seqValueBefore = await ReadSequenceLastValueAsync(Schema, $"mt_events_sequence_{tenant}"); seqValueBefore.ShouldBeGreaterThanOrEqualTo(2L); - await _store.Advanced.RemoveMartenManagedTenantsAsync(new[] { tenant }, CancellationToken.None); + await Store.Advanced.RemoveMartenManagedTenantsAsync(new[] { tenant }, CancellationToken.None); // Partition table dropped. - (await TableExistsAsync(_schema, $"mt_events_{tenant}")).ShouldBeFalse(); + (await TableExistsAsync(Schema, $"mt_events_{tenant}")).ShouldBeFalse(); // #4683: sequence is dropped too (was the second orphan-leak pin). - (await SequenceExistsAsync(_schema, $"mt_events_sequence_{tenant}")).ShouldBeFalse( + (await SequenceExistsAsync(Schema, $"mt_events_sequence_{tenant}")).ShouldBeFalse( "RemoveMartenManagedTenantsAsync now drops the per-tenant mt_events_sequence_ " + "via PerTenantPartitionedCleanup (#4683). Was previously pinned as the orphan leak."); } @@ -142,12 +118,12 @@ public async Task DeleteAllTenantDataAsync_removes_per_tenant_progression_rows_a // the HighWaterShardIdentity grammar) and leaves store-global rows alone. var keep = "keepme"; var drop = "dropme"; - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, keep, drop); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, keep, drop); // Touch the events table so the cleaner's batched DELETEs find data to delete (the // partition drop itself is the load-bearing part of this test; the actual event count // is incidental). - await using (var session = _store.LightweightSession(drop)) + await using (var session = Store.LightweightSession(drop)) { session.Events.StartStream(Guid.NewGuid(), new DelEvent("x")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -159,7 +135,7 @@ public async Task DeleteAllTenantDataAsync_removes_per_tenant_progression_rows_a // * {Name}:V2:All: — versioned variant // Plus a store-global "HighWaterMark" row + a "SomeProjection:All" row that must // survive the drop. - await SeedProgressionRowsAsync(_schema, new[] + await SeedProgressionRowsAsync(Schema, new[] { // store-global -- must survive "HighWaterMark", @@ -174,15 +150,15 @@ await SeedProgressionRowsAsync(_schema, new[] $"VersionedProjection:V2:All:{drop}", }); - var beforeNames = await ReadProgressionRowNamesAsync(_schema); + var beforeNames = await ReadProgressionRowNamesAsync(Schema); beforeNames.ShouldContain($"HighWaterMark:{drop}"); beforeNames.ShouldContain($"DelCountProjection:All:{drop}"); beforeNames.ShouldContain($"VersionedProjection:V2:All:{drop}"); // Act. - await _store.Advanced.DeleteAllTenantDataAsync(drop, CancellationToken.None); + await Store.Advanced.DeleteAllTenantDataAsync(drop, CancellationToken.None); - var afterNames = await ReadProgressionRowNamesAsync(_schema); + var afterNames = await ReadProgressionRowNamesAsync(Schema); // The dropped tenant's per-tenant rows are gone, across both grammars + the versioned form. afterNames.Any(n => MentionsTenant(n, drop)).ShouldBeFalse( diff --git a/src/TenantPartitionedEventsTests/AppendWrite/event_metadata_propagation_under_partitioning.cs b/src/TenantPartitionedEventsTests/AppendWrite/event_metadata_propagation_under_partitioning.cs index b3e9a5fa70..0fdf630f83 100644 --- a/src/TenantPartitionedEventsTests/AppendWrite/event_metadata_propagation_under_partitioning.cs +++ b/src/TenantPartitionedEventsTests/AppendWrite/event_metadata_propagation_under_partitioning.cs @@ -79,55 +79,34 @@ public async Task event_TenantId_equals_stream_TenantId_under_partitioning() /// values propagate onto every event the bulk function inserts — and that they /// stay paired with the right tenant_id in the partition. /// -public class event_optional_metadata_propagation_under_partitioning : IAsyncLifetime +public class event_optional_metadata_propagation_under_partitioning : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_meta"; - public async ValueTask InitializeAsync() - { - _schema = $"tp_meta_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); - - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - // Opt in to the four opt-in metadata columns. - opts.Events.MetadataConfig.CausationIdEnabled = true; - opts.Events.MetadataConfig.CorrelationIdEnabled = true; - opts.Events.MetadataConfig.HeadersEnabled = true; - opts.Events.MetadataConfig.UserNameEnabled = true; - - opts.Events.AddEventType(); - }); - } + protected override bool EnsureStorageOnInitialize => false; - public ValueTask DisposeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _store?.Dispose(); - return default; + // Opt in to the four opt-in metadata columns. + opts.Events.MetadataConfig.CausationIdEnabled = true; + opts.Events.MetadataConfig.CorrelationIdEnabled = true; + opts.Events.MetadataConfig.HeadersEnabled = true; + opts.Events.MetadataConfig.UserNameEnabled = true; + + opts.Events.AddEventType(); } [Fact] public async Task session_metadata_propagates_to_each_event_in_the_tenants_partition() { - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); var correlation = "corr-" + Guid.NewGuid().ToString("N")[..10]; var causation = "caus-" + Guid.NewGuid().ToString("N")[..10]; var userName = "user-" + Guid.NewGuid().ToString("N")[..8]; var streamId = Guid.NewGuid(); - await using (var s = _store.LightweightSession("alpha")) + await using (var s = Store.LightweightSession("alpha")) { s.CorrelationId = correlation; s.CausationId = causation; @@ -138,7 +117,7 @@ public async Task session_metadata_propagates_to_each_event_in_the_tenants_parti await s.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using var q = _store.QuerySession("alpha"); + await using var q = Store.QuerySession("alpha"); var events = await q.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); events.Count.ShouldBe(3); diff --git a/src/TenantPartitionedEventsTests/Daemon/dead_letter_under_partitioning.cs b/src/TenantPartitionedEventsTests/Daemon/dead_letter_under_partitioning.cs index 131766a618..fce827ecaa 100644 --- a/src/TenantPartitionedEventsTests/Daemon/dead_letter_under_partitioning.cs +++ b/src/TenantPartitionedEventsTests/Daemon/dead_letter_under_partitioning.cs @@ -12,6 +12,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -43,41 +44,16 @@ namespace TenantPartitionedEventsTests.Daemon; /// in the schema) to be meaningful. /// /// -public class dead_letter_under_partitioning : IAsyncLifetime +public class dead_letter_under_partitioning : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_dlq"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_dlq_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); - - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - // Touch the events feature so MartenManagedTenantListPartitions - // runs its applyPartitioning sweep over every doc mapping at - // schema-feature build time. - opts.Events.AddEventType(); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + // Touch the events feature so MartenManagedTenantListPartitions + // runs its applyPartitioning sweep over every doc mapping at + // schema-feature build time. + opts.Events.AddEventType(); } [Fact] @@ -88,7 +64,7 @@ public void dead_letter_event_mapping_stays_single_tenanted_under_partitioning() // that auto-multi-tenants every mapping (or removes the explicit // carve-out in MartenManagedTenantListPartitions / StoreOptions) is a // deliberate contract change. - _store.StorageFeatures.MappingFor(typeof(DeadLetterEvent)).TenancyStyle + Store.StorageFeatures.MappingFor(typeof(DeadLetterEvent)).TenancyStyle .ShouldBe(TenancyStyle.Single, "DeadLetterEvent is store-global by design — daemon diagnostics aren't a tenant boundary"); } @@ -100,8 +76,8 @@ public void dead_letter_event_lives_under_events_schema_not_doc_schema() // Events.DatabaseSchemaName (not the default doc schema). Pin the // schema placement so a future refactor that moves it doesn't silently // strand existing dead-letter rows. - var mapping = _store.StorageFeatures.MappingFor(typeof(DeadLetterEvent)); - mapping.DatabaseSchemaName.ShouldBe(_store.Options.Events.DatabaseSchemaName); + var mapping = Store.StorageFeatures.MappingFor(typeof(DeadLetterEvent)); + mapping.DatabaseSchemaName.ShouldBe(Store.Options.Events.DatabaseSchemaName); } [Fact] @@ -115,12 +91,12 @@ public void dead_letter_event_table_has_no_tenant_id_partition_declaration_under // dead-letter consumer, and (b) silo dead-letter rows per tenant when // the contract is store-global. Pin by querying the live partition // map for the dead-letter table — it must NOT be registered. - var dlqTable = _store.Options.TenantPartitions?.Partitions; + var dlqTable = Store.Options.TenantPartitions?.Partitions; if (dlqTable == null) return; // no partition manager active — nothing to pin // Pull the underlying partitioned-table set; the dead-letter table // must not appear there. - var allPartitionedTableNames = _store.Storage.AllObjects() + var allPartitionedTableNames = Store.Storage.AllObjects() .OfType() .Where(t => t.Partitioning is Weasel.Postgresql.Tables.Partitioning.ListPartitioning) .Select(t => t.Identifier.Name) diff --git a/src/TenantPartitionedEventsTests/Daemon/per_tenant_rebuild_cancellation.cs b/src/TenantPartitionedEventsTests/Daemon/per_tenant_rebuild_cancellation.cs index 5ae3a2ece9..f80dac1b86 100644 --- a/src/TenantPartitionedEventsTests/Daemon/per_tenant_rebuild_cancellation.cs +++ b/src/TenantPartitionedEventsTests/Daemon/per_tenant_rebuild_cancellation.cs @@ -35,51 +35,35 @@ namespace TenantPartitionedEventsTests.Daemon; /// no drain loops. /// /// -public class per_tenant_rebuild_cancellation: IAsyncLifetime +public class per_tenant_rebuild_cancellation: PartitionedStoreContext { - private static readonly string SchemaName = $"rebuild_cancel_{Environment.ProcessId}"; + protected override string SchemaPrefix => "rebuild_cancel"; - private DocumentStore _store = null!; + protected override string BuildSchemaName() => $"rebuild_cancel_{Environment.ProcessId}"; - public async ValueTask InitializeAsync() - { - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = SchemaName; - opts.AutoCreateSchemaObjects = AutoCreate.All; - - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - // Unique advisory-lock id so this store's daemon machinery never contends - // with the shared partitioned fixtures running in sibling collections. - opts.Projections.DaemonLockId = 4791; + protected override bool DropSchemaOnInitialize => false; - opts.Projections.Add(new GatedPerEventProjection(), ProjectionLifecycle.Async, - GatedPerEventProjection.ProjectionName); - opts.Schema.For().DocumentAlias("cancel_tally"); - }); + protected override void ConfigureStore(StoreOptions opts) + { + opts.AutoCreateSchemaObjects = AutoCreate.All; - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } + // Unique advisory-lock id so this store's daemon machinery never contends + // with the shared partitioned fixtures running in sibling collections. + opts.Projections.DaemonLockId = 4791; - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Projections.Add(new GatedPerEventProjection(), ProjectionLifecycle.Async, + GatedPerEventProjection.ProjectionName); + opts.Schema.For().DocumentAlias("cancel_tally"); } [Fact] public async Task cancelling_a_per_tenant_rebuild_leaves_progression_consistent_and_the_cell_rebuildable() { var tenant = PartitionedFixtureBase.NewTenant(); - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); const int eventCount = 40; - await using (var session = _store.LightweightSession(tenant)) + await using (var session = Store.LightweightSession(tenant)) { for (var i = 0; i < 4; i++) { @@ -97,7 +81,7 @@ public async Task cancelling_a_per_tenant_rebuild_leaves_progression_consistent_ GatedPerEventProjection.Gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using var cts = new CancellationTokenSource(); - using var daemon = await _store.BuildProjectionDaemonAsync(); + using var daemon = await Store.BuildProjectionDaemonAsync(); var rebuildTask = daemon.RebuildProjectionAsync( GatedPerEventProjection.ProjectionName, tenant, cts.Token); @@ -136,7 +120,7 @@ public async Task cancelling_a_per_tenant_rebuild_leaves_progression_consistent_ await daemon.RebuildProjectionAsync(GatedPerEventProjection.ProjectionName, tenant, CancellationToken.None); - await using (var query = _store.QuerySession(tenant)) + await using (var query = Store.QuerySession(tenant)) { (await query.Query().CountAsync(TestContext.Current.CancellationToken)).ShouldBe(eventCount, "the follow-up rebuild must fully materialize the cell"); @@ -147,9 +131,9 @@ await daemon.RebuildProjectionAsync(GatedPerEventProjection.ProjectionName, tena public async Task pre_cancelled_token_does_not_disturb_the_cell() { var tenant = PartitionedFixtureBase.NewTenant(); - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); - await using (var session = _store.LightweightSession(tenant)) + await using (var session = Store.LightweightSession(tenant)) { session.Events.StartStream(Guid.NewGuid(), new TallyEvent(), new TallyEvent()); await session.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -158,7 +142,7 @@ public async Task pre_cancelled_token_does_not_disturb_the_cell() GatedPerEventProjection.Gate = null; GatedPerEventProjection.Started = null; - using var daemon = await _store.BuildProjectionDaemonAsync(); + using var daemon = await Store.BuildProjectionDaemonAsync(); using var cancelled = new CancellationTokenSource(); cancelled.Cancel(); @@ -176,7 +160,7 @@ await daemon.RebuildProjectionAsync(GatedPerEventProjection.ProjectionName, tena await daemon.RebuildProjectionAsync(GatedPerEventProjection.ProjectionName, tenant, CancellationToken.None); - await using var query = _store.QuerySession(tenant); + await using var query = Store.QuerySession(tenant); (await query.Query().CountAsync(TestContext.Current.CancellationToken)).ShouldBe(2); } @@ -186,7 +170,7 @@ private async Task> ReadCellProgressionsAsync(string tenantI await conn.OpenAsync(); await using var cmd = conn.CreateCommand(); cmd.CommandText = - $"select last_seq_id from {SchemaName}.mt_event_progression where name like @name and name like @tenant"; + $"select last_seq_id from {Schema}.mt_event_progression where name like @name and name like @tenant"; cmd.Parameters.AddWithValue("name", GatedPerEventProjection.ProjectionName + "%"); cmd.Parameters.AddWithValue("tenant", "%" + tenantId + "%"); diff --git a/src/TenantPartitionedEventsTests/Dcb/dcb_cross_tenant_query_isolation_pin.cs b/src/TenantPartitionedEventsTests/Dcb/dcb_cross_tenant_query_isolation_pin.cs index e68cefe117..e14a7c265e 100644 --- a/src/TenantPartitionedEventsTests/Dcb/dcb_cross_tenant_query_isolation_pin.cs +++ b/src/TenantPartitionedEventsTests/Dcb/dcb_cross_tenant_query_isolation_pin.cs @@ -12,6 +12,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -48,46 +49,19 @@ namespace TenantPartitionedEventsTests.Dcb; /// TagTables to keep the pin focused on the path that was broken. /// /// -public class dcb_cross_tenant_query_isolation_pin : IAsyncLifetime +public class dcb_cross_tenant_query_isolation_pin : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_dcbx"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_dcbx_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + // TagTables is the broken-JOIN path. HStore stores the tag on the + // event row itself so the cross-tenant query is intrinsically + // safe; this pin is specifically about the JOIN under TagTables. + opts.Events.DcbStorageMode = DcbStorageMode.TagTables; - await using (var conn = new NpgsqlConnection(ConnectionSource.ConnectionString)) - { - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - // TagTables is the broken-JOIN path. HStore stores the tag on the - // event row itself so the cross-tenant query is intrinsically - // safe; this pin is specifically about the JOIN under TagTables. - opts.Events.DcbStorageMode = DcbStorageMode.TagTables; - - opts.Events.AddEventType(); - opts.Events.RegisterTagType("dcbxt_customer"); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Events.AddEventType(); + opts.Events.RegisterTagType("dcbxt_customer"); } [Fact] @@ -101,11 +75,11 @@ public async Task tag_query_under_partitioning_returns_only_own_tenant_event() // event — the #4645 fix tightened the JOIN to also match on tenant_id // so the per-tenant-seq-id collision no longer produces a duplicate // (or — if the WHERE clause hadn't held — a cross-tenant leak). - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); var sharedCustomer = new DcbXtCustomerId(Guid.NewGuid()); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { var evt = session.Events.BuildEvent(new DcbXtPayment("ALPHA-PAYMENT")); evt.WithTag(sharedCustomer); @@ -113,7 +87,7 @@ public async Task tag_query_under_partitioning_returns_only_own_tenant_event() await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { var evt = session.Events.BuildEvent(new DcbXtPayment("BETA-PAYMENT")); evt.WithTag(sharedCustomer); @@ -123,7 +97,7 @@ public async Task tag_query_under_partitioning_returns_only_own_tenant_event() var query = new EventTagQuery().Or(sharedCustomer); - await using var alphaQuery = _store.LightweightSession("alpha"); + await using var alphaQuery = Store.LightweightSession("alpha"); var events = await alphaQuery.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); // Exactly one row — alpha's own event. The fix to EventStore.Dcb.cs: diff --git a/src/TenantPartitionedEventsTests/Dcb/dcb_tag_append_under_partitioning.cs b/src/TenantPartitionedEventsTests/Dcb/dcb_tag_append_under_partitioning.cs index dd1cd2bab1..bfa26c3b73 100644 --- a/src/TenantPartitionedEventsTests/Dcb/dcb_tag_append_under_partitioning.cs +++ b/src/TenantPartitionedEventsTests/Dcb/dcb_tag_append_under_partitioning.cs @@ -10,6 +10,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -43,47 +44,17 @@ namespace TenantPartitionedEventsTests.Dcb; /// all under partitioning. /// /// -public class dcb_tag_append_under_partitioning : IAsyncLifetime +public class dcb_tag_append_under_partitioning : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_dcb"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - // Same own-store schema convention as Bug_4611 — Guid + ProcessId fits - // under PG's 32-char comfort threshold for nested partition + sequence - // suffix names. - _schema = $"tp_dcb_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); - - await using (var conn = new NpgsqlConnection(ConnectionSource.ConnectionString)) - { - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - // One string-keyed tag — the simplest shape that exercises the - // tag side-table column types (text) + the tenant_id PK column - // EventTagTable adds under conjoined tenancy. - opts.Events.RegisterTagType("dcb_order_ref"); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Events.AddEventType(); + // One string-keyed tag — the simplest shape that exercises the + // tag side-table column types (text) + the tenant_id PK column + // EventTagTable adds under conjoined tenancy. + opts.Events.RegisterTagType("dcb_order_ref"); } [Fact] @@ -96,11 +67,11 @@ public async Task dcb_tag_append_succeeds_under_partitioning_per_tenant() // operation runs as a sibling Marten storage operation in the same // session, so it must continue to work under the per-tenant // partitioned append path. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); var orderRef = new DcbOrderRef("ORD-" + Guid.NewGuid().ToString("N")[..8]); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { var evt = session.Events.BuildEvent(new DcbOrderPlaced("widget")); evt.WithTag(orderRef); @@ -115,7 +86,7 @@ public async Task dcb_tag_append_succeeds_under_partitioning_per_tenant() await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); await conn.OpenAsync(TestContext.Current.CancellationToken); await using var cmd = conn.CreateCommand( - $"select count(*) from {_schema}.mt_event_tag_dcb_order_ref where value = :v and tenant_id = :t"); + $"select count(*) from {Schema}.mt_event_tag_dcb_order_ref where value = :v and tenant_id = :t"); cmd.Parameters.AddWithValue("v", orderRef.Value); cmd.Parameters.AddWithValue("t", "alpha"); var count = (long)(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken))!; @@ -143,11 +114,11 @@ public async Task dcb_tag_query_round_trips_per_tenant() // added as well. Filed as a follow-up — out of scope for this test // file, which only covers the append + same-tenant read happy paths // requested by #4617 section 3f. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); var orderRef = new DcbOrderRef("ORD-rt-" + Guid.NewGuid().ToString("N")[..8]); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { var evt = session.Events.BuildEvent(new DcbOrderPlaced("alpha-widget")); evt.WithTag(orderRef); @@ -157,7 +128,7 @@ public async Task dcb_tag_query_round_trips_per_tenant() var query = new EventTagQuery().Or(orderRef); - await using (var queryA = _store.LightweightSession("alpha")) + await using (var queryA = Store.LightweightSession("alpha")) { var events = await queryA.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); events.Count.ShouldBe(1, diff --git a/src/TenantPartitionedEventsTests/Fixtures/PartitionedStoreContext.cs b/src/TenantPartitionedEventsTests/Fixtures/PartitionedStoreContext.cs new file mode 100644 index 0000000000..c61c82fa8d --- /dev/null +++ b/src/TenantPartitionedEventsTests/Fixtures/PartitionedStoreContext.cs @@ -0,0 +1,122 @@ +#nullable enable +using System; +using System.Threading.Tasks; +using JasperFx.Events; +using JasperFx.MultiTenancy; +using Marten; +using Marten.Events; +using Marten.Testing.Harness; +using Npgsql; +using Weasel.Postgresql; +using Xunit; + +namespace TenantPartitionedEventsTests.Fixtures; + +/// +/// Per-test-class store lifecycle for tests that need their OWN +/// under UseTenantPartitionedEvents — the +/// cases where registering file-local projections or flipping a store-level +/// flag on the shared fixtures would +/// pollute sibling tests. Absorbs the copy-paste template that used to be +/// re-declared per file: drop a fresh pid+guid schema, build the store with +/// the common partitioned-tenancy config, ensure event storage, dispose. +/// +/// +/// Subclasses override for file-local event +/// types, projections, and flag deviations — it runs AFTER the common config +/// lines so it can override any of them. Prefer the shared collection +/// fixtures when a test doesn't need store-level customization; per-test +/// isolation there is by unique tenant id, which is much cheaper than a +/// schema per class. +/// +public abstract class PartitionedStoreContext: IAsyncLifetime +{ + protected DocumentStore Store { get; private set; } = null!; + protected string Schema { get; private set; } = null!; + + /// + /// Short lowercase schema tag, e.g. tp_dlq. Keep it terse — the + /// generated name {prefix}_{pid}_{guid} is truncated to 32 chars + /// and must stay unique-per-run after truncation. + /// + protected abstract string SchemaPrefix { get; } + + /// + /// File-local store configuration. Runs after the common partitioned + /// config (Conjoined + UseTenantPartitionedEvents + + /// QuickWithServerTimestamps + AllDocumentsAreMultiTenanted), so it can + /// override any of those lines. + /// + protected abstract void ConfigureStore(StoreOptions opts); + + /// + /// Lowercase, hyphen-free, leading-digit-safe (#4567), under PostgreSQL's + /// identifier limit with room for Marten's table suffixes. Override for + /// tests pinned to a stable schema across store rebuilds. + /// + protected virtual string BuildSchemaName() + { + var name = $"{SchemaPrefix}_{Environment.ProcessId}_{Guid.NewGuid():N}"; + return name.Length <= 32 ? name : name.Substring(0, 32); + } + + /// + /// Override to false for tests that intentionally build against whatever + /// schema state is already present. + /// + protected virtual bool DropSchemaOnInitialize => true; + + /// + /// Override to false for tests that must observe the store BEFORE event + /// storage exists (e.g. first-touch schema creation behavior). + /// + protected virtual bool EnsureStorageOnInitialize => true; + + public virtual async ValueTask InitializeAsync() + { + await BuildFreshStoreAsync(); + } + + /// + /// Disposes any current store, mints a fresh schema name, and rebuilds. + /// Re-entrant on purpose: concurrency-regression tests re-run the whole + /// lifecycle per attempt. + /// + protected async Task BuildFreshStoreAsync() + { + Store?.Dispose(); + + Schema = BuildSchemaName(); + + if (DropSchemaOnInitialize) + { + await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync(); + try { await conn.DropSchemaAsync(Schema); } catch { } + } + + Store = DocumentStore.For(opts => + { + opts.Connection(ConnectionSource.ConnectionString); + opts.DatabaseSchemaName = Schema; + + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + opts.Events.UseTenantPartitionedEvents = true; + opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; + opts.Policies.AllDocumentsAreMultiTenanted(); + + ConfigureStore(opts); + }); + + if (EnsureStorageOnInitialize) + { + await Store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); + } + } + + public virtual ValueTask DisposeAsync() + { + Store?.Dispose(); + return default; + } +} diff --git a/src/TenantPartitionedEventsTests/Projections/add_global_projection_under_partitioning.cs b/src/TenantPartitionedEventsTests/Projections/add_global_projection_under_partitioning.cs index 5b98c395f9..1bd665f43a 100644 --- a/src/TenantPartitionedEventsTests/Projections/add_global_projection_under_partitioning.cs +++ b/src/TenantPartitionedEventsTests/Projections/add_global_projection_under_partitioning.cs @@ -13,6 +13,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -30,44 +31,19 @@ namespace TenantPartitionedEventsTests.Projections; /// the AllDocumentsAreMultiTenanted policy. /// /// -public class add_global_projection_under_partitioning : IAsyncLifetime +public class add_global_projection_under_partitioning : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_glob"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_glob_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + opts.Events.AddEventType(); - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - - // The headline: register a SingleStreamProjection as GLOBAL — its - // aggregate doc keeps TenancyStyle.Single even though the source - // event store is partitioned per tenant and the default policy - // multi-tenants every other doc. - opts.Projections.AddGlobalProjection(new GlobalCounterProjection(), ProjectionLifecycle.Inline); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + // The headline: register a SingleStreamProjection as GLOBAL — its + // aggregate doc keeps TenancyStyle.Single even though the source + // event store is partitioned per tenant and the default policy + // multi-tenants every other doc. + opts.Projections.AddGlobalProjection(new GlobalCounterProjection(), ProjectionLifecycle.Inline); } [Fact] @@ -77,7 +53,7 @@ public void global_projection_doc_is_TenancyStyle_Single_under_partitioning() // the global projection's aggregate doc stays Single-tenanted by design. // Pin so a future change that auto-multi-tenants the global doc is a // deliberate contract change. - _store.StorageFeatures.MappingFor(typeof(GlobalCounter)).TenancyStyle + Store.StorageFeatures.MappingFor(typeof(GlobalCounter)).TenancyStyle .ShouldBe(TenancyStyle.Single); } @@ -93,19 +69,19 @@ public async Task append_to_global_aggregate_succeeds_and_rolls_up_across_tenant // store has global aggregates registered. The rerouted appends then land // in a real partition (with their own per-tenant event sequence) instead // of raising MT002. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); var globalId = Guid.NewGuid(); // Two different tenants funnel events into the SAME global stream - await using (var alpha = _store.LightweightSession("alpha")) + await using (var alpha = Store.LightweightSession("alpha")) { alpha.Events.StartStream(globalId, new GlobalTickEvent("first"), new GlobalTickEvent("second")); await alpha.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var beta = _store.LightweightSession("beta")) + await using (var beta = Store.LightweightSession("beta")) { beta.Events.Append(globalId, new GlobalTickEvent("third")); await beta.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -113,7 +89,7 @@ public async Task append_to_global_aggregate_succeeds_and_rolls_up_across_tenant // The inline global projection doc is single-tenanted, so it reads the // same from any tenant's session — and reflects BOTH tenants' appends - await using (var reader = _store.QuerySession("beta")) + await using (var reader = Store.QuerySession("beta")) { var counter = await reader.LoadAsync(globalId, TestContext.Current.CancellationToken); counter.ShouldNotBeNull(); @@ -126,12 +102,12 @@ public async Task append_to_global_aggregate_succeeds_and_rolls_up_across_tenant await conn.OpenAsync(TestContext.Current.CancellationToken); var suffix = (string?)await conn.CreateCommand( - $"select partition_suffix from {_schema}.mt_tenant_partitions where partition_value = '{StorageConstants.DefaultTenantId}'") + $"select partition_suffix from {Schema}.mt_tenant_partitions where partition_value = '{StorageConstants.DefaultTenantId}'") .ExecuteScalarAsync(TestContext.Current.CancellationToken); suffix.ShouldBe("__default__"); var eventCount = (long)(await conn.CreateCommand( - $"select count(*) from {_schema}.mt_events___default__") + $"select count(*) from {Schema}.mt_events___default__") .ExecuteScalarAsync(TestContext.Current.CancellationToken))!; eventCount.ShouldBe(3); } @@ -143,7 +119,7 @@ public async Task reserved_default_suffix_is_rejected_for_regular_tenants() // global-projection default tenant slot — a shared suffix would fold two // partition VALUES into one partition table and corrupt tenant isolation. var ex = await Should.ThrowAsync(() => - _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, + Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, new System.Collections.Generic.Dictionary { ["acme"] = "__default__" })); ex.Message.ShouldContain("reserved"); diff --git a/src/TenantPartitionedEventsTests/Projections/custom_aggregate_grouper_per_tenant.cs b/src/TenantPartitionedEventsTests/Projections/custom_aggregate_grouper_per_tenant.cs index 6bd4bd996f..7f3feb998b 100644 --- a/src/TenantPartitionedEventsTests/Projections/custom_aggregate_grouper_per_tenant.cs +++ b/src/TenantPartitionedEventsTests/Projections/custom_aggregate_grouper_per_tenant.cs @@ -17,6 +17,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -46,47 +47,22 @@ namespace TenantPartitionedEventsTests.Projections; /// would change the shared fixture's projection set for every sibling test. /// /// -public class custom_aggregate_grouper_per_tenant : IAsyncLifetime +public class custom_aggregate_grouper_per_tenant : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_grouper"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_grouper_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); - - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - opts.Events.AddEventType(); - opts.Events.AddEventType(); - - // Inline lookup: external-account-id -> customer-id (string-keyed - // since ExternalAccountId is the natural identity). - opts.Projections.Add(ProjectionLifecycle.Inline); - // Inline billing: uses the custom grouper to fan ShippingLabelCreated - // out to the matching customer via the link table. - opts.Projections.Add(ProjectionLifecycle.Inline); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Events.AddEventType(); + opts.Events.AddEventType(); + opts.Events.AddEventType(); + + // Inline lookup: external-account-id -> customer-id (string-keyed + // since ExternalAccountId is the natural identity). + opts.Projections.Add(ProjectionLifecycle.Inline); + // Inline billing: uses the custom grouper to fan ShippingLabelCreated + // out to the matching customer via the link table. + opts.Projections.Add(ProjectionLifecycle.Inline); } [Fact] @@ -98,20 +74,20 @@ public async Task grouper_lookup_resolves_only_to_same_tenant_link_docs() // beta's link doc and route beta's shipping events to alpha's customer // (or vice versa). The pin: each tenant's billing metrics reflect ONLY // its own customer's shipping events. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); const string sharedExternalId = "ACME-PRO"; var alphaCustomer = Guid.NewGuid(); var betaCustomer = Guid.NewGuid(); // alpha: register customer, link to ACME-PRO, send 3 shipping labels. - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(alphaCustomer, new CustomerRegistered(alphaCustomer, "Alpha Inc")); session.Events.Append(alphaCustomer, new CustomerLinkedToExternalAccount(alphaCustomer, sharedExternalId)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { var labelStream = Guid.NewGuid(); session.Events.StartStream(labelStream, @@ -123,13 +99,13 @@ public async Task grouper_lookup_resolves_only_to_same_tenant_link_docs() // beta: register a DIFFERENT customer, link to the SAME external id, // send 5 shipping labels. - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { session.Events.StartStream(betaCustomer, new CustomerRegistered(betaCustomer, "Beta LLC")); session.Events.Append(betaCustomer, new CustomerLinkedToExternalAccount(betaCustomer, sharedExternalId)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { var labelStream = Guid.NewGuid(); session.Events.StartStream(labelStream, @@ -144,13 +120,13 @@ public async Task grouper_lookup_resolves_only_to_same_tenant_link_docs() // Read each tenant's billing doc using a tenant-scoped query. The // billing metric is multi-tenanted (per AllDocumentsAreMultiTenanted), // so each tenant sees only its own row keyed by its own customer id. - await using var alphaQuery = _store.QuerySession("alpha"); + await using var alphaQuery = Store.QuerySession("alpha"); var alphaBilling = await alphaQuery.LoadAsync(alphaCustomer, TestContext.Current.CancellationToken); alphaBilling.ShouldNotBeNull("alpha's grouper must have routed alpha's labels to alpha's customer"); alphaBilling!.ShippingLabels.ShouldBe(3, "alpha appended 3 labels — beta's 5 must not bleed in via shared external id"); - await using var betaQuery = _store.QuerySession("beta"); + await using var betaQuery = Store.QuerySession("beta"); var betaBilling = await betaQuery.LoadAsync(betaCustomer, TestContext.Current.CancellationToken); betaBilling.ShouldNotBeNull("beta's grouper must have routed beta's labels to beta's customer"); betaBilling!.ShippingLabels.ShouldBe(5, @@ -163,16 +139,16 @@ public async Task tenant_A_billing_doc_invisible_from_tenant_B_session() // Sibling pin: the billing doc itself is partitioned per tenant (it's // a multi-tenanted doc). Querying tenant B with tenant A's customer id // must return null — no doc visibility across tenant slots. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); var alphaCustomer = Guid.NewGuid(); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(alphaCustomer, new CustomerRegistered(alphaCustomer, "Alpha Inc")); session.Events.Append(alphaCustomer, new CustomerLinkedToExternalAccount(alphaCustomer, "EXT-A")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { var labelStream = Guid.NewGuid(); session.Events.StartStream(labelStream, new ShippingLabelCreated("EXT-A")); @@ -180,14 +156,14 @@ public async Task tenant_A_billing_doc_invisible_from_tenant_B_session() } // Pin from alpha's own session: doc exists. - await using (var alphaQuery = _store.QuerySession("alpha")) + await using (var alphaQuery = Store.QuerySession("alpha")) { (await alphaQuery.LoadAsync(alphaCustomer, TestContext.Current.CancellationToken)).ShouldNotBeNull(); } // Pin from beta's session: alpha's customer id finds nothing — the // billing doc is in alpha's tenant slot, beta's tenant slot is empty. - await using (var betaQuery = _store.QuerySession("beta")) + await using (var betaQuery = Store.QuerySession("beta")) { (await betaQuery.LoadAsync(alphaCustomer, TestContext.Current.CancellationToken)) .ShouldBeNull("alpha's billing doc must not be visible to beta — tenant slot isolation"); diff --git a/src/TenantPartitionedEventsTests/Projections/determine_action_async_per_tenant.cs b/src/TenantPartitionedEventsTests/Projections/determine_action_async_per_tenant.cs index 6755e24ea1..70262ca941 100644 --- a/src/TenantPartitionedEventsTests/Projections/determine_action_async_per_tenant.cs +++ b/src/TenantPartitionedEventsTests/Projections/determine_action_async_per_tenant.cs @@ -15,6 +15,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -42,45 +43,25 @@ namespace TenantPartitionedEventsTests.Projections; /// projection set. /// /// -public class determine_action_async_per_tenant : IAsyncLifetime +public class determine_action_async_per_tenant : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_da"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_da_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + opts.Events.AddEventType(); + opts.Events.AddEventType(); - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } + // Inline so the projection runs in the writing session. + opts.Projections.Add(ProjectionLifecycle.Inline); + } + public override async ValueTask InitializeAsync() + { // Reset cross-test capture so each test starts clean. DetermineCounterProjection.ObservedTenants.Clear(); - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - opts.Events.AddEventType(); - - // Inline so the projection runs in the writing session. - opts.Projections.Add(ProjectionLifecycle.Inline); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + await base.InitializeAsync(); } [Fact] @@ -90,15 +71,15 @@ public async Task determine_action_async_observes_writing_tenant_id_per_slice() // captures session.TenantId on every call. Pin: the set of observed // tenant ids matches the writing tenants exactly — no *DEFAULT*, no // cross-tenant leak. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(Guid.NewGuid(), new DetermineIncrementEvent(), new DetermineIncrementEvent()); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { session.Events.StartStream(Guid.NewGuid(), new DetermineIncrementEvent()); @@ -128,18 +109,18 @@ public async Task delete_via_determine_action_scoped_to_writing_tenant_only() // each tenant's slot has its own stream). alpha sends one // DetermineResetEvent which returns ActionType.Delete. beta's doc // for the same stream id must NOT be deleted. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); var sharedStream = Guid.NewGuid(); // alpha: start + increment once → doc with Count = 1. - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(sharedStream, new DetermineIncrementEvent()); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } // beta: start + increment twice → doc with Count = 2. - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { session.Events.StartStream(sharedStream, new DetermineIncrementEvent(), new DetermineIncrementEvent()); @@ -147,30 +128,30 @@ public async Task delete_via_determine_action_scoped_to_writing_tenant_only() } // Confirm both docs exist with their counts. - await using (var alphaQuery = _store.QuerySession("alpha")) + await using (var alphaQuery = Store.QuerySession("alpha")) { (await alphaQuery.LoadAsync(sharedStream, TestContext.Current.CancellationToken))!.Count.ShouldBe(1); } - await using (var betaQuery = _store.QuerySession("beta")) + await using (var betaQuery = Store.QuerySession("beta")) { (await betaQuery.LoadAsync(sharedStream, TestContext.Current.CancellationToken))!.Count.ShouldBe(2); } // alpha sends a DetermineResetEvent → projection returns Delete → // alpha's doc is removed. - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.Append(sharedStream, new DetermineResetEvent()); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } // Pin: alpha's doc is gone, beta's is untouched (Count still 2). - await using (var alphaQuery = _store.QuerySession("alpha")) + await using (var alphaQuery = Store.QuerySession("alpha")) { (await alphaQuery.LoadAsync(sharedStream, TestContext.Current.CancellationToken)) .ShouldBeNull("alpha's reset → Delete should have removed alpha's doc"); } - await using (var betaQuery = _store.QuerySession("beta")) + await using (var betaQuery = Store.QuerySession("beta")) { var betaDoc = await betaQuery.LoadAsync(sharedStream, TestContext.Current.CancellationToken); betaDoc.ShouldNotBeNull("beta's doc must NOT be deleted by alpha's reset (tenant isolation)"); diff --git a/src/TenantPartitionedEventsTests/Projections/event_projection_per_tenant.cs b/src/TenantPartitionedEventsTests/Projections/event_projection_per_tenant.cs index 593bd0d86e..535e7658af 100644 --- a/src/TenantPartitionedEventsTests/Projections/event_projection_per_tenant.cs +++ b/src/TenantPartitionedEventsTests/Projections/event_projection_per_tenant.cs @@ -13,6 +13,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -34,41 +35,16 @@ namespace TenantPartitionedEventsTests.Projections; /// future fixture refactor cannot drift the assertions. /// /// -public class event_projection_per_tenant : IAsyncLifetime +public class event_projection_per_tenant : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_evproj"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_evproj_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + opts.Events.AddEventType(); + opts.Schema.For().Identity(x => x.Id).DocumentAlias("p2c_leg_log"); - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - opts.Schema.For().Identity(x => x.Id).DocumentAlias("p2c_leg_log"); - - opts.Projections.Add(ProjectionLifecycle.Inline); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Projections.Add(ProjectionLifecycle.Inline); } [Fact] @@ -79,23 +55,23 @@ public async Task event_projection_inline_materializes_only_in_owning_tenants_se // in alpha's session; beta (no events appended) sees no docs. var alpha = "alpha_" + Guid.NewGuid().ToString("N")[..8]; var beta = "beta_" + Guid.NewGuid().ToString("N")[..8]; - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, alpha, beta); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, alpha, beta); var alphaStream = Guid.NewGuid(); - await using (var session = _store.LightweightSession(alpha)) + await using (var session = Store.LightweightSession(alpha)) { session.Events.StartStream(alphaStream, new LegEvent(1.0), new LegEvent(2.0), new LegEvent(3.0)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using var qa = _store.QuerySession(alpha); + await using var qa = Store.QuerySession(alpha); var alphaLogs = await qa.Query().ToListAsync(TestContext.Current.CancellationToken); alphaLogs.Count.ShouldBe(3, "EventProjection.Create emits one LegLog per LegEvent — alpha appended 3, so alpha's session sees 3"); alphaLogs.Select(l => l.Distance).OrderBy(d => d).ShouldBe(new[] { 1.0, 2.0, 3.0 }); - await using var qb = _store.QuerySession(beta); + await using var qb = Store.QuerySession(beta); var betaLogs = await qb.Query().ToListAsync(TestContext.Current.CancellationToken); betaLogs.ShouldBeEmpty( "beta appended no events — its tenant slot must be empty (no cross-tenant doc leak)"); @@ -111,44 +87,44 @@ public async Task event_projection_async_via_RebuildProjectionAsync_materializes // byte-identical (never touched by the per-tenant rebuild). var alpha = "alpha_" + Guid.NewGuid().ToString("N")[..8]; var beta = "beta_" + Guid.NewGuid().ToString("N")[..8]; - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, alpha, beta); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, alpha, beta); var alphaStream = Guid.NewGuid(); var betaStream = Guid.NewGuid(); - await using (var session = _store.LightweightSession(alpha)) + await using (var session = Store.LightweightSession(alpha)) { session.Events.StartStream(alphaStream, new LegEvent(10), new LegEvent(20)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession(beta)) + await using (var session = Store.LightweightSession(beta)) { session.Events.StartStream(betaStream, new LegEvent(7), new LegEvent(13)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } // Pre-state: inline already materialized both tenants. - await using (var qa = _store.QuerySession(alpha)) + await using (var qa = Store.QuerySession(alpha)) { (await qa.Query().ToListAsync(TestContext.Current.CancellationToken)).Count.ShouldBe(2); } - await using (var qb = _store.QuerySession(beta)) + await using (var qb = Store.QuerySession(beta)) { (await qb.Query().ToListAsync(TestContext.Current.CancellationToken)).Count.ShouldBe(2); } // Wipe alpha's docs + progression — Phase 2c teardown. Beta untouched. - var es = (IEventStore)_store; + var es = (IEventStore)Store; await es.DeleteProjectionProgressAsync( - (IEventDatabase)_store.Storage.Database, + (IEventDatabase)Store.Storage.Database, LegLoggerProjection.ProjectionName, tenantId: alpha, CancellationToken.None); - await using (var qa = _store.QuerySession(alpha)) + await using (var qa = Store.QuerySession(alpha)) { (await qa.Query().ToListAsync(TestContext.Current.CancellationToken)) .ShouldBeEmpty("alpha's docs must be wiped by the tenant-scoped teardown"); } - await using (var qb = _store.QuerySession(beta)) + await using (var qb = Store.QuerySession(beta)) { (await qb.Query().ToListAsync(TestContext.Current.CancellationToken)).Count.ShouldBe(2, "beta's docs must survive — the teardown is tenant-scoped to alpha"); @@ -156,20 +132,20 @@ await es.DeleteProjectionProgressAsync( // Rebuild ONLY alpha. The per-tenant overload routes the rebuild to a // tenant-scoped shard; beta is not touched. - using (var daemon = await _store.BuildProjectionDaemonAsync()) + using (var daemon = await Store.BuildProjectionDaemonAsync()) { await daemon.RebuildProjectionAsync( LegLoggerProjection.ProjectionName, alpha, CancellationToken.None); } - await using (var qa = _store.QuerySession(alpha)) + await using (var qa = Store.QuerySession(alpha)) { var alphaLogs = await qa.Query().ToListAsync(TestContext.Current.CancellationToken); alphaLogs.Count.ShouldBe(2, "alpha was rebuilt — its 2 LegEvents must materialize 2 LegLog docs"); alphaLogs.Select(l => l.Distance).OrderBy(d => d).ShouldBe(new[] { 10.0, 20.0 }); } - await using (var qb = _store.QuerySession(beta)) + await using (var qb = Store.QuerySession(beta)) { (await qb.Query().ToListAsync(TestContext.Current.CancellationToken)).Count.ShouldBe(2, "beta's docs must STILL be 2 — the per-tenant rebuild for alpha did not touch beta"); diff --git a/src/TenantPartitionedEventsTests/Projections/flat_table_projection_per_tenant.cs b/src/TenantPartitionedEventsTests/Projections/flat_table_projection_per_tenant.cs index 952933260d..834dd92de2 100644 --- a/src/TenantPartitionedEventsTests/Projections/flat_table_projection_per_tenant.cs +++ b/src/TenantPartitionedEventsTests/Projections/flat_table_projection_per_tenant.cs @@ -12,6 +12,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -44,42 +45,17 @@ namespace TenantPartitionedEventsTests.Projections; /// set. /// /// -public class flat_table_projection_per_tenant : IAsyncLifetime +public class flat_table_projection_per_tenant : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_flat"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_flat_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + opts.Events.AddEventType(); - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - - // Inline so writes from a tenant session land synchronously in - // the flat table — keeps the assertions deterministic. - opts.Projections.Add(new FtCounterProjection(), ProjectionLifecycle.Inline); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + // Inline so writes from a tenant session land synchronously in + // the flat table — keeps the assertions deterministic. + opts.Projections.Add(new FtCounterProjection(), ProjectionLifecycle.Inline); } [Fact] @@ -90,14 +66,14 @@ public async Task ddl_applies_cleanly_under_partitioning() // its DDL to apply. With UseTenantPartitionedEvents on, no FK / // partition error should appear (related to the #4606 carve-out // that dropped the explicit mt_events → mt_streams FK). - await _store.Storage.ApplyAllConfiguredChangesToDatabaseAsync(); + await Store.Storage.ApplyAllConfiguredChangesToDatabaseAsync(); await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); await conn.OpenAsync(TestContext.Current.CancellationToken); await using var cmd = conn.CreateCommand(); cmd.CommandText = "select count(*) from information_schema.tables where table_schema = @s and table_name = 'ft_counters'"; - cmd.Parameters.AddWithValue("s", _schema); + cmd.Parameters.AddWithValue("s", Schema); var count = (long)(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken))!; count.ShouldBe(1L, "FlatTableProjection's ft_counters table must exist after schema apply"); } @@ -108,18 +84,18 @@ public async Task per_tenant_inline_writes_land_in_flat_table_for_unique_pks() // Each tenant uses a DIFFERENT counterId — proves the inline write path // works end-to-end under partitioning. (Same-PK across tenants is the // collision pin below.) - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); var alphaCounter = Guid.NewGuid(); var betaCounter = Guid.NewGuid(); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(alphaCounter, new FtCounterIncremented(alphaCounter, 10)); session.Events.Append(alphaCounter, new FtCounterIncremented(alphaCounter, 5)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { session.Events.StartStream(betaCounter, new FtCounterIncremented(betaCounter, 100)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -148,12 +124,12 @@ public async Task cross_tenant_same_pk_silently_overwrites_pin_user_managed_isol // // Pinned so a future change that auto-tenants flat tables flips // this assertion intentionally. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); var sharedCounterId = Guid.NewGuid(); // alpha writes first. - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(sharedCounterId, new FtCounterIncremented(sharedCounterId, 10)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -163,7 +139,7 @@ public async Task cross_tenant_same_pk_silently_overwrites_pin_user_managed_isol // beta increments the SAME row (silent overwrite of the row owner — // the framework happily applies beta's session writes to the same PK). - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { session.Events.StartStream(sharedCounterId, new FtCounterIncremented(sharedCounterId, 100)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -184,7 +160,7 @@ private async Task ReadTotalAsync(Guid counterId) await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); await conn.OpenAsync(); await using var cmd = conn.CreateCommand(); - cmd.CommandText = $"select total from {_schema}.ft_counters where id = @id"; + cmd.CommandText = $"select total from {Schema}.ft_counters where id = @id"; cmd.Parameters.AddWithValue("id", counterId); var raw = await cmd.ExecuteScalarAsync(); return raw == null || raw is DBNull ? 0 : (int)raw; diff --git a/src/TenantPartitionedEventsTests/Projections/multi_stream_projection_rollup_by_tenant.cs b/src/TenantPartitionedEventsTests/Projections/multi_stream_projection_rollup_by_tenant.cs index 086a1411d7..d2fcd71065 100644 --- a/src/TenantPartitionedEventsTests/Projections/multi_stream_projection_rollup_by_tenant.cs +++ b/src/TenantPartitionedEventsTests/Projections/multi_stream_projection_rollup_by_tenant.cs @@ -14,6 +14,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -47,45 +48,20 @@ namespace TenantPartitionedEventsTests.Projections; /// looked up. /// /// -public class multi_stream_projection_rollup_by_tenant : IAsyncLifetime +public class multi_stream_projection_rollup_by_tenant : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_rollup"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_rollup_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + opts.Events.AddEventType(); + opts.Events.AddEventType(); - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } + // Short alias keeps the partition / index identifiers under PG's + // 64-byte limit when combined with tenant slot names downstream. + opts.Schema.For().DocumentAlias("p2c_tenant_rollup"); - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - opts.Events.AddEventType(); - - // Short alias keeps the partition / index identifiers under PG's - // 64-byte limit when combined with tenant slot names downstream. - opts.Schema.For().DocumentAlias("p2c_tenant_rollup"); - - opts.Projections.Add(ProjectionLifecycle.Async); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Projections.Add(ProjectionLifecycle.Async); } [Fact] @@ -97,9 +73,9 @@ public async Task RollUpByTenant_produces_one_doc_per_tenant_aggregating_their_e // that tenant's events. var alpha = "alpha_" + Guid.NewGuid().ToString("N")[..8]; var beta = "beta_" + Guid.NewGuid().ToString("N")[..8]; - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, alpha, beta); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, alpha, beta); - await using (var session = _store.LightweightSession(alpha)) + await using (var session = Store.LightweightSession(alpha)) { var acct = Guid.NewGuid(); session.Events.StartStream(acct, @@ -108,7 +84,7 @@ public async Task RollUpByTenant_produces_one_doc_per_tenant_aggregating_their_e new TransactionPosted(acct, 50m)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession(beta)) + await using (var session = Store.LightweightSession(beta)) { var acct = Guid.NewGuid(); session.Events.StartStream(acct, @@ -124,7 +100,7 @@ public async Task RollUpByTenant_produces_one_doc_per_tenant_aggregating_their_e // since the rollup grouping turns the per-tenant data into single docs // by tenant id and we want determinism on the assertion, not race-y // wait-for-completion semantics. - using (var daemon = await _store.BuildProjectionDaemonAsync()) + using (var daemon = await Store.BuildProjectionDaemonAsync()) { await daemon.RebuildProjectionAsync(TenantRollupProjection.ProjectionName, alpha, CancellationToken.None); await daemon.RebuildProjectionAsync(TenantRollupProjection.ProjectionName, beta, CancellationToken.None); @@ -133,7 +109,7 @@ public async Task RollUpByTenant_produces_one_doc_per_tenant_aggregating_their_e // The rollup doc lives in the default tenant slot (slicer hardcodes // DefaultTenantId as the group's TenantId) — read from a default-tenant // session keyed by the doc identity (= tenant id string). - await using var query = _store.QuerySession(); + await using var query = Store.QuerySession(); var alphaRollup = await query.LoadAsync(alpha, TestContext.Current.CancellationToken); var betaRollup = await query.LoadAsync(beta, TestContext.Current.CancellationToken); @@ -154,10 +130,10 @@ public async Task RollUpByTenant_doc_for_tenant_A_only_aggregates_As_events() // rollup doc totals are exactly that tenant's events, not the union. var alpha = "alpha_" + Guid.NewGuid().ToString("N")[..8]; var beta = "beta_" + Guid.NewGuid().ToString("N")[..8]; - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, alpha, beta); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, alpha, beta); var alphaAcct = Guid.NewGuid(); - await using (var session = _store.LightweightSession(alpha)) + await using (var session = Store.LightweightSession(alpha)) { session.Events.StartStream(alphaAcct, new AccountOpened(alphaAcct)); session.Events.Append(alphaAcct, @@ -168,7 +144,7 @@ public async Task RollUpByTenant_doc_for_tenant_A_only_aggregates_As_events() } var betaAcct = Guid.NewGuid(); - await using (var session = _store.LightweightSession(beta)) + await using (var session = Store.LightweightSession(beta)) { session.Events.StartStream(betaAcct, new AccountOpened(betaAcct)); session.Events.Append(betaAcct, @@ -177,13 +153,13 @@ public async Task RollUpByTenant_doc_for_tenant_A_only_aggregates_As_events() await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - using (var daemon = await _store.BuildProjectionDaemonAsync()) + using (var daemon = await Store.BuildProjectionDaemonAsync()) { await daemon.RebuildProjectionAsync(TenantRollupProjection.ProjectionName, alpha, CancellationToken.None); await daemon.RebuildProjectionAsync(TenantRollupProjection.ProjectionName, beta, CancellationToken.None); } - await using var query = _store.QuerySession(); + await using var query = Store.QuerySession(); var alphaRollup = await query.LoadAsync(alpha, TestContext.Current.CancellationToken); var betaRollup = await query.LoadAsync(beta, TestContext.Current.CancellationToken); diff --git a/src/TenantPartitionedEventsTests/Projections/raw_iprojection_per_tenant.cs b/src/TenantPartitionedEventsTests/Projections/raw_iprojection_per_tenant.cs index 075bde52ec..9e8c9eaf77 100644 --- a/src/TenantPartitionedEventsTests/Projections/raw_iprojection_per_tenant.cs +++ b/src/TenantPartitionedEventsTests/Projections/raw_iprojection_per_tenant.cs @@ -15,6 +15,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -42,48 +43,28 @@ namespace TenantPartitionedEventsTests.Projections; /// projection registration. /// /// -public class raw_iprojection_per_tenant : IAsyncLifetime +public class raw_iprojection_per_tenant : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_rawproj"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_rawproj_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + opts.Events.AddEventType(); - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } + // Inline so the projection runs in the same session that's writing + // the events — the tightest pin we can make for "the IDocumentOperations + // passed in is tenant-scoped". + opts.Projections.Add(new TenantTouchProjection(), ProjectionLifecycle.Inline); + } + public override async ValueTask InitializeAsync() + { // Reset cross-test capture before each store stands up so assertions // on TenantIdsSeen don't pick up sibling test data. TenantTouchProjection.TenantIdsSeen.Clear(); TenantTouchProjection.BatchTenantIds.Clear(); - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - - // Inline so the projection runs in the same session that's writing - // the events — the tightest pin we can make for "the IDocumentOperations - // passed in is tenant-scoped". - opts.Projections.Add(new TenantTouchProjection(), ProjectionLifecycle.Inline); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + await base.InitializeAsync(); } [Fact] @@ -94,10 +75,10 @@ public async Task raw_iprojection_writes_doc_to_correct_tenant_slot_per_batch() // must be readable from that tenant's QuerySession and INVISIBLE from // the sibling tenant's session — proves IDocumentOperations.Store is // tenant-scoped under partitioning. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); var alphaStream = Guid.NewGuid(); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(alphaStream, new TenantTouchEvent("alpha-1"), new TenantTouchEvent("alpha-2")); @@ -105,20 +86,20 @@ public async Task raw_iprojection_writes_doc_to_correct_tenant_slot_per_batch() } var betaStream = Guid.NewGuid(); - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { session.Events.StartStream(betaStream, new TenantTouchEvent("beta-1")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } // Each tenant's doc IS visible from its own session. - await using (var alphaQuery = _store.QuerySession("alpha")) + await using (var alphaQuery = Store.QuerySession("alpha")) { var alphaDoc = await alphaQuery.LoadAsync(alphaStream, TestContext.Current.CancellationToken); alphaDoc.ShouldNotBeNull(); alphaDoc!.TouchCount.ShouldBe(2); } - await using (var betaQuery = _store.QuerySession("beta")) + await using (var betaQuery = Store.QuerySession("beta")) { var betaDoc = await betaQuery.LoadAsync(betaStream, TestContext.Current.CancellationToken); betaDoc.ShouldNotBeNull(); @@ -127,12 +108,12 @@ public async Task raw_iprojection_writes_doc_to_correct_tenant_slot_per_batch() // Cross-tenant load returns null — proves the doc is in the writing // tenant's slot, not bleeding into the other. - await using (var alphaQuery = _store.QuerySession("alpha")) + await using (var alphaQuery = Store.QuerySession("alpha")) { (await alphaQuery.LoadAsync(betaStream, TestContext.Current.CancellationToken)) .ShouldBeNull("beta's doc must not be visible to alpha"); } - await using (var betaQuery = _store.QuerySession("beta")) + await using (var betaQuery = Store.QuerySession("beta")) { (await betaQuery.LoadAsync(alphaStream, TestContext.Current.CancellationToken)) .ShouldBeNull("alpha's doc must not be visible to beta"); @@ -146,14 +127,14 @@ public async Task each_inline_call_sees_events_from_exactly_one_tenant() // is called inline, every event in the batch carries the SAME TenantId // (the writing session's tenant). Cross-tenant fan-in does not happen // at the inline call site — that's an async-daemon-only concern. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(Guid.NewGuid(), new TenantTouchEvent("alpha-call")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { session.Events.StartStream(Guid.NewGuid(), new TenantTouchEvent("beta-call")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); diff --git a/src/TenantPartitionedEventsTests/Projections/same_projection_lifecycle_equivalence.cs b/src/TenantPartitionedEventsTests/Projections/same_projection_lifecycle_equivalence.cs index ae81c40993..08c759d042 100644 --- a/src/TenantPartitionedEventsTests/Projections/same_projection_lifecycle_equivalence.cs +++ b/src/TenantPartitionedEventsTests/Projections/same_projection_lifecycle_equivalence.cs @@ -13,6 +13,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -33,47 +34,22 @@ namespace TenantPartitionedEventsTests.Projections; /// Local copies of the event + aggregate keep the test self-contained. /// /// -public class same_projection_lifecycle_equivalence : IAsyncLifetime +public class same_projection_lifecycle_equivalence : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_lifecycle"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_lifecycle_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); - - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - opts.Events.AddEventType(); - - // Inline materializes a TripSummary per stream as events are - // appended. Live aggregation reads the stream + folds via Apply() - // on demand — same Apply() body, so any drift between the two - // codepaths surfaces as a non-equal aggregate. - opts.Schema.For().Identity(x => x.Id).DocumentAlias("p2c_trip_summary"); - opts.Projections.Add(ProjectionLifecycle.Inline); - opts.Projections.LiveStreamAggregation(); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Events.AddEventType(); + opts.Events.AddEventType(); + + // Inline materializes a TripSummary per stream as events are + // appended. Live aggregation reads the stream + folds via Apply() + // on demand — same Apply() body, so any drift between the two + // codepaths surfaces as a non-equal aggregate. + opts.Schema.For().Identity(x => x.Id).DocumentAlias("p2c_trip_summary"); + opts.Projections.Add(ProjectionLifecycle.Inline); + opts.Projections.LiveStreamAggregation(); } [Fact] @@ -83,10 +59,10 @@ public async Task Inline_and_Live_projections_yield_identical_per_tenant_aggrega // events. Read the inline-materialized doc and the live-aggregated // result; pin Distance + LegCount byte-equal between the two paths. var tenant = "alpha_" + Guid.NewGuid().ToString("N")[..8]; - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); var streamId = Guid.NewGuid(); - await using (var session = _store.LightweightSession(tenant)) + await using (var session = Store.LightweightSession(tenant)) { session.Events.StartStream(streamId, new TripStartedV2(streamId), @@ -96,7 +72,7 @@ public async Task Inline_and_Live_projections_yield_identical_per_tenant_aggrega await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using var query = _store.QuerySession(tenant); + await using var query = Store.QuerySession(tenant); // Inline-materialized doc: read directly from the projected document // table (no folding at read time). diff --git a/src/TenantPartitionedEventsTests/ReadQuery/enable_unique_index_on_event_id_under_partitioning.cs b/src/TenantPartitionedEventsTests/ReadQuery/enable_unique_index_on_event_id_under_partitioning.cs index 5f28b85452..a609b8b3f6 100644 --- a/src/TenantPartitionedEventsTests/ReadQuery/enable_unique_index_on_event_id_under_partitioning.cs +++ b/src/TenantPartitionedEventsTests/ReadQuery/enable_unique_index_on_event_id_under_partitioning.cs @@ -12,6 +12,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -37,39 +38,15 @@ namespace TenantPartitionedEventsTests.ReadQuery; /// — flipping it on the shared fixture would affect every sibling test. /// /// -public class enable_unique_index_on_event_id_under_partitioning : IAsyncLifetime +public class enable_unique_index_on_event_id_under_partitioning : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_eui"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_eui_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + opts.Events.EnableUniqueIndexOnEventId = true; - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Events.EnableUniqueIndexOnEventId = true; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Events.AddEventType(); } [Fact] @@ -79,13 +56,13 @@ public async Task same_event_id_in_two_tenants_does_NOT_violate_unique_index_und // (local per partition). The same Guid event id can land in two // tenants' partitions without violation — each partition's local // index sees only its own rows. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha", "beta"); var sharedEventId = Guid.NewGuid(); var alphaStream = Guid.NewGuid(); var betaStream = Guid.NewGuid(); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(alphaStream, new Event(new UniqEvent("a")) { Id = sharedEventId }); await session.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -93,7 +70,7 @@ public async Task same_event_id_in_two_tenants_does_NOT_violate_unique_index_und // Beta appends an event with the SAME id — should succeed because the // unique index is scoped per tenant partition. - await using (var session = _store.LightweightSession("beta")) + await using (var session = Store.LightweightSession("beta")) { session.Events.StartStream(betaStream, new Event(new UniqEvent("b")) { Id = sharedEventId }); await session.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -103,7 +80,7 @@ public async Task same_event_id_in_two_tenants_does_NOT_violate_unique_index_und await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); await conn.OpenAsync(TestContext.Current.CancellationToken); await using var cmd = conn.CreateCommand( - $"select count(*) from {_schema}.mt_events where id = @id"); + $"select count(*) from {Schema}.mt_events where id = @id"); cmd.Parameters.AddWithValue("id", sharedEventId); var count = (long)(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken))!; count.ShouldBe(2L, diff --git a/src/TenantPartitionedEventsTests/Regressions/Bug_4596_partition_race_and_MT002_rebuild.cs b/src/TenantPartitionedEventsTests/Regressions/Bug_4596_partition_race_and_MT002_rebuild.cs index a99ca02979..9601246453 100644 --- a/src/TenantPartitionedEventsTests/Regressions/Bug_4596_partition_race_and_MT002_rebuild.cs +++ b/src/TenantPartitionedEventsTests/Regressions/Bug_4596_partition_race_and_MT002_rebuild.cs @@ -43,40 +43,14 @@ namespace TenantPartitionedEventsTests.Regressions; /// could mask the other's pin on a shared fixture. /// /// -public class Bug_4596_partition_race_and_MT002_rebuild : IAsyncLifetime +public class Bug_4596_partition_race_and_MT002_rebuild : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_4596"; - public ValueTask InitializeAsync() => new ValueTask(freshStoreAsync()); - - // Build a brand-new isolated store against a brand-new schema. Used by InitializeAsync and re-used - // per attempt by the concurrent-race test so each retry starts from a clean slate. - private async Task freshStoreAsync() + protected override void ConfigureStore(StoreOptions opts) { - _store?.Dispose(); - _schema = $"tp_4596_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); - - await using (var conn = new NpgsqlConnection(ConnectionSource.ConnectionString)) - { - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - opts.Events.AddEventType(); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); + opts.Events.AddEventType(); + opts.Events.AddEventType(); } // The race in this test occasionally trips a KNOWN, unrelated transient in Weasel's concurrent @@ -89,12 +63,6 @@ private async Task freshStoreAsync() private static bool IsKnownWeaselConcurrentReadTransient(Exception e) => e is NullReferenceException && (e.StackTrace?.Contains("readExistingAsync") ?? false); - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; - } - [Fact] public async Task concurrent_AddMartenManagedTenantsAsync_for_same_tenant_is_idempotent() { @@ -115,10 +83,10 @@ public async Task concurrent_AddMartenManagedTenantsAsync_for_same_tenant_is_ide var attempt = 1; while (true) { - await freshStoreAsync(); + await BuildFreshStoreAsync(); tasks = Enumerable.Range(0, 3) - .Select(_ => Task.Run(() => _store.Advanced.AddMartenManagedTenantsAsync( + .Select(_ => Task.Run(() => Store.Advanced.AddMartenManagedTenantsAsync( CancellationToken.None, raceyTenant))) .ToArray(); @@ -158,7 +126,7 @@ public async Task concurrent_AddMartenManagedTenantsAsync_for_same_tenant_is_ide await using (var cmd = conn.CreateCommand( "select count(*) from information_schema.tables where table_schema = :s and table_name = :n")) { - cmd.Parameters.AddWithValue("s", _schema); + cmd.Parameters.AddWithValue("s", Schema); cmd.Parameters.AddWithValue("n", $"mt_events_{raceyTenant}"); var partitionCount = (long)(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken))!; partitionCount.ShouldBe(1L, @@ -169,7 +137,7 @@ public async Task concurrent_AddMartenManagedTenantsAsync_for_same_tenant_is_ide await using (var cmd = conn.CreateCommand( "select count(*) from information_schema.sequences where sequence_schema = :s and sequence_name = :n")) { - cmd.Parameters.AddWithValue("s", _schema); + cmd.Parameters.AddWithValue("s", Schema); cmd.Parameters.AddWithValue("n", $"mt_events_sequence_{raceyTenant}"); var seqCount = (long)(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken))!; seqCount.ShouldBe(1L, @@ -180,7 +148,7 @@ public async Task concurrent_AddMartenManagedTenantsAsync_for_same_tenant_is_ide // append — the partition + sequence + tenant-registration row are // all coherent. Pre-fix, a losing task occasionally left a // partial registration that made the FIRST append fire MT002. - await using (var session = _store.LightweightSession(raceyTenant)) + await using (var session = Store.LightweightSession(raceyTenant)) { session.Events.StartStream( Guid.NewGuid(), new Bug4596TripStarted("post-race append")); @@ -197,9 +165,9 @@ public async Task RebuildProjectionAsync_for_unregistered_tenant_is_empty_noop_N // filtered by tenant_id) — it never calls mt_quick_append_events, // so MT002 ("Tenant '...' has no registered partition") doesn't fire. // The rebuild just finds zero events for that tenant and is a no-op. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "registered_alpha"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "registered_alpha"); - await using (var session = _store.LightweightSession("registered_alpha")) + await using (var session = Store.LightweightSession("registered_alpha")) { session.Events.StartStream( Guid.NewGuid(), @@ -213,7 +181,7 @@ public async Task RebuildProjectionAsync_for_unregistered_tenant_is_empty_noop_N // silently create a partition under the user's feet). The check is // wrapped in a try so the exception type assertion is unambiguous. Exception? appendException = null; - await using (var session = _store.LightweightSession("typo_tenant")) + await using (var session = Store.LightweightSession("typo_tenant")) { session.Events.StartStream( Guid.NewGuid(), new Bug4596TripStarted("should-fail")); @@ -233,7 +201,7 @@ public async Task RebuildProjectionAsync_for_unregistered_tenant_is_empty_noop_N // Now the complementary pin: rebuild for the same unregistered // tenant must NOT throw MT002 (or anything else) — it walks zero // events via EventLoader's SELECT path and returns cleanly. - using (var daemon = await _store.BuildProjectionDaemonAsync()) + using (var daemon = await Store.BuildProjectionDaemonAsync()) { // No projection registered on this store — but the call shape // still goes through the same daemon entry point that walks the @@ -251,7 +219,7 @@ public async Task RebuildProjectionAsync_for_unregistered_tenant_is_empty_noop_N await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); await conn.OpenAsync(TestContext.Current.CancellationToken); await using var cmd = conn.CreateCommand( - $"select count(*) from {_schema}.mt_events where tenant_id = :t"); + $"select count(*) from {Schema}.mt_events where tenant_id = :t"); cmd.Parameters.AddWithValue("t", "typo_tenant"); var unregisteredCount = (long)(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken))!; unregisteredCount.ShouldBe(0L, diff --git a/src/TenantPartitionedEventsTests/Regressions/Bug_4611_mandatory_stream_type_under_partitioning.cs b/src/TenantPartitionedEventsTests/Regressions/Bug_4611_mandatory_stream_type_under_partitioning.cs index 31a0c61dae..d782290801 100644 --- a/src/TenantPartitionedEventsTests/Regressions/Bug_4611_mandatory_stream_type_under_partitioning.cs +++ b/src/TenantPartitionedEventsTests/Regressions/Bug_4611_mandatory_stream_type_under_partitioning.cs @@ -11,6 +11,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -41,39 +42,15 @@ namespace TenantPartitionedEventsTests.Regressions; /// affect every sibling test on a shared fixture. /// /// -public class Bug_4611_mandatory_stream_type_under_partitioning : IAsyncLifetime +public class Bug_4611_mandatory_stream_type_under_partitioning : PartitionedStoreContext { - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_4611"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_4611_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32); + opts.Events.UseMandatoryStreamTypeDeclaration = true; - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Events.UseMandatoryStreamTypeDeclaration = true; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Events.AddEventType(); } [Fact] @@ -81,27 +58,27 @@ public async Task StartStream_with_aggregate_type_followed_by_Append_works() { // The headline #4611 regression — Start then Append must both succeed, // and the mt_streams row's type column must carry the AggregateType name. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); var streamId = Guid.NewGuid(); - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.StartStream(streamId, new MandatoryEvent("first")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var query = _store.QuerySession("alpha")) + await using (var query = Store.QuerySession("alpha")) { (await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken)).Count.ShouldBe(1); } - await using (var session = _store.LightweightSession("alpha")) + await using (var session = Store.LightweightSession("alpha")) { session.Events.Append(streamId, new MandatoryEvent("second")); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } - await using (var query = _store.QuerySession("alpha")) + await using (var query = Store.QuerySession("alpha")) { (await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken)).Count.ShouldBe(2); } @@ -109,12 +86,12 @@ public async Task StartStream_with_aggregate_type_followed_by_Append_works() // The end state pin: mt_streams row exists with type = aggregate type alias. await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); await conn.OpenAsync(TestContext.Current.CancellationToken); - await using var cmd = conn.CreateCommand($"select type from {_schema}.mt_streams where id = @id and tenant_id = @tid"); + await using var cmd = conn.CreateCommand($"select type from {Schema}.mt_streams where id = @id and tenant_id = @tid"); cmd.Parameters.AddWithValue("id", streamId); cmd.Parameters.AddWithValue("tid", "alpha"); var typeName = (string?)await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken); typeName.ShouldNotBeNull("mt_streams row must exist (not tombstoned) — the headline #4611 regression"); - typeName.ShouldBe(_store.Events.AggregateAliasFor(typeof(MandatoryAggregate))); + typeName.ShouldBe(Store.Events.AggregateAliasFor(typeof(MandatoryAggregate))); } [Fact] @@ -125,9 +102,9 @@ public async Task untyped_StartStream_still_throws_StreamTypeMissingException() // UseMandatoryStreamTypeDeclaration is on. Pre-#4613's fix the // bulk-path post-process guard was the wrong layer; the correct guard // (this one) was never the problem. - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, "alpha"); - await using var session = _store.LightweightSession("alpha"); + await using var session = Store.LightweightSession("alpha"); Should.Throw(() => { diff --git a/src/TenantPartitionedEventsTests/Regressions/schema_update_preserves_existing_per_tenant_sequences.cs b/src/TenantPartitionedEventsTests/Regressions/schema_update_preserves_existing_per_tenant_sequences.cs index d2c1909bd3..a1b827047a 100644 --- a/src/TenantPartitionedEventsTests/Regressions/schema_update_preserves_existing_per_tenant_sequences.cs +++ b/src/TenantPartitionedEventsTests/Regressions/schema_update_preserves_existing_per_tenant_sequences.cs @@ -9,6 +9,7 @@ using Marten.Testing.Harness; using Npgsql; using Shouldly; +using TenantPartitionedEventsTests.Fixtures; using Weasel.Postgresql; using Xunit; @@ -26,53 +27,26 @@ namespace TenantPartitionedEventsTests.Regressions; /// back before damage was done; with idempotent partition DDL (weasel#326) that accidental guard is /// gone, so the delta itself must be additive-only. /// -public class schema_update_preserves_existing_per_tenant_sequences : IAsyncLifetime +public class schema_update_preserves_existing_per_tenant_sequences : PartitionedStoreContext { private const string TenantA = "tenant_a"; private const string TenantB = "tenant_b"; - private string _schema = null!; - private DocumentStore _store = null!; + protected override string SchemaPrefix => "tp_seqpreserve"; - public async ValueTask InitializeAsync() + protected override void ConfigureStore(StoreOptions opts) { - _schema = $"tp_seqpreserve_{Guid.NewGuid():N}".Substring(0, 32); - - await using (var conn = new NpgsqlConnection(ConnectionSource.ConnectionString)) - { - await conn.OpenAsync(); - try { await conn.DropSchemaAsync(_schema); } catch { } - } - - _store = DocumentStore.For(opts => - { - opts.Connection(ConnectionSource.ConnectionString); - opts.DatabaseSchemaName = _schema; - opts.Events.TenancyStyle = TenancyStyle.Conjoined; - opts.Events.UseTenantPartitionedEvents = true; - opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; - opts.Policies.AllDocumentsAreMultiTenanted(); - - opts.Events.AddEventType(); - }); - - await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent)); - } - - public ValueTask DisposeAsync() - { - _store?.Dispose(); - return default; + opts.Events.AddEventType(); } [Fact] public async Task adding_a_missing_sequence_does_not_reset_existing_ones() { - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, TenantA); - await _store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, TenantB); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, TenantA); + await Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, TenantB); // Advance tenant A's event sequence by appending real events. - await using (var session = _store.LightweightSession(TenantA)) + await using (var session = Store.LightweightSession(TenantA)) { session.Events.StartStream(Guid.NewGuid(), new SeqPreserveEvent(1), new SeqPreserveEvent(2)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); @@ -82,41 +56,41 @@ public async Task adding_a_missing_sequence_does_not_reset_existing_ones() await conn.OpenAsync(TestContext.Current.CancellationToken); var valueBefore = (long)(await conn - .CreateCommand($"select last_value from \"{_schema}\".\"mt_events_sequence_{TenantA}\"") + .CreateCommand($"select last_value from \"{Schema}\".\"mt_events_sequence_{TenantA}\"") .ExecuteScalarAsync(TestContext.Current.CancellationToken))!; valueBefore.ShouldBeGreaterThanOrEqualTo(2); // Simulate the canary scenario: a tenant is registered on the shared partition list but its // sequence is missing in this database (mid-provisioning / a racing applier's stale snapshot), // so the next schema apply computes an Update delta for the per-tenant sequences. - await conn.CreateCommand($"drop sequence \"{_schema}\".\"mt_events_sequence_{TenantB}\"") + await conn.CreateCommand($"drop sequence \"{Schema}\".\"mt_events_sequence_{TenantB}\"") .ExecuteNonQueryAsync(TestContext.Current.CancellationToken); - await _store.Storage.Database.ApplyAllConfiguredChangesToDatabaseAsync(ct: TestContext.Current.CancellationToken); + await Store.Storage.Database.ApplyAllConfiguredChangesToDatabaseAsync(ct: TestContext.Current.CancellationToken); // The missing sequence is (re)created... var tenantBExists = await conn .CreateCommand( "select count(*) from information_schema.sequences " + - $"where sequence_schema = '{_schema}' and sequence_name = 'mt_events_sequence_{TenantB}'") + $"where sequence_schema = '{Schema}' and sequence_name = 'mt_events_sequence_{TenantB}'") .ExecuteScalarAsync(TestContext.Current.CancellationToken); tenantBExists.ShouldBe(1L); // ...and tenant A's sequence kept its value: the update was additive-only, no drop+recreate. var valueAfter = (long)(await conn - .CreateCommand($"select last_value from \"{_schema}\".\"mt_events_sequence_{TenantA}\"") + .CreateCommand($"select last_value from \"{Schema}\".\"mt_events_sequence_{TenantA}\"") .ExecuteScalarAsync(TestContext.Current.CancellationToken))!; valueAfter.ShouldBe(valueBefore); // And the next append continues from where the sequence left off instead of colliding at 1. - await using (var session = _store.LightweightSession(TenantA)) + await using (var session = Store.LightweightSession(TenantA)) { session.Events.StartStream(Guid.NewGuid(), new SeqPreserveEvent(3)); await session.SaveChangesAsync(TestContext.Current.CancellationToken); } var valueAfterAppend = (long)(await conn - .CreateCommand($"select last_value from \"{_schema}\".\"mt_events_sequence_{TenantA}\"") + .CreateCommand($"select last_value from \"{Schema}\".\"mt_events_sequence_{TenantA}\"") .ExecuteScalarAsync(TestContext.Current.CancellationToken))!; valueAfterAppend.ShouldBeGreaterThan(valueBefore); }