Skip to content

Absorb the hand-rolled IAsyncLifetime template in TenantPartitionedEventsTests #5100

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

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

Absorb the hand-rolled IAsyncLifetime template in TenantPartitionedEventsTests #5100

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

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 13 additions & 36 deletions src/TenantPartitionedEventsTests/Admin/admin_extras_under_partitioning.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
using Marten.Testing.Harness;
using Npgsql;
using Shouldly;
using TenantPartitionedEventsTests.Fixtures;
using Weasel.Postgresql;
using Xunit;

Expand All @@ -31,36 +32,12 @@ namespace TenantPartitionedEventsTests.Admin;
/// sequences after registering N tenants; re-apply is idempotent.</item>
/// </list>
/// </summary>
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]
Expand All @@ -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");
Expand All @@ -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]
Expand All @@ -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");
}
Expand Down
76 changes: 26 additions & 50 deletions src/TenantPartitionedEventsTests/Admin/delete_all_tenant_data_orphan_sequence_pin.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
using Marten.Testing.Harness;
using Npgsql;
using Shouldly;
using TenantPartitionedEventsTests.Fixtures;
using Weasel.Postgresql;
using Xunit;

Expand All @@ -29,71 +30,46 @@ namespace TenantPartitionedEventsTests.Admin;
/// <see cref="HighWaterShardIdentity.PerTenantPrefix"/> match) is the dropped tenant.
/// Store-global progression rows are intentionally left alone.
/// </summary>
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<DelEvent>();
// #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<DelCountProjection>(ProjectionLifecycle.Async);
});

await _store.Storage.Database.EnsureStorageExistsAsync(typeof(IEvent));
}

public ValueTask DisposeAsync()
{
_store?.Dispose();
return default;
opts.Events.AddEventType<DelEvent>();
// #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<DelCountProjection>(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_<tenant> via " +
"PerTenantPartitionedCleanup (#4683). Was previously pinned as the orphan leak.");
Expand All @@ -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_<tenant> " +
"via PerTenantPartitionedCleanup (#4683). Was previously pinned as the orphan leak.");
}
Expand All @@ -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);
Expand All @@ -159,7 +135,7 @@ public async Task DeleteAllTenantDataAsync_removes_per_tenant_progression_rows_a
// * {Name}:V2:All:<tenant> — 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",
Expand All @@ -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(
Expand Down
49 changes: 14 additions & 35 deletions ...TenantPartitionedEventsTests/AppendWrite/event_metadata_propagation_under_partitioning.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
/// </summary>
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<MetaEvent>();
});
}
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<MetaEvent>();
}

[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;
Expand All @@ -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);

Expand Down
Loading
Loading