Skip to content

Migrate raw-DocumentStore.For event sourcing tests onto standard harnesses #5099

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
+22 −96
Merged

Migrate raw-DocumentStore.For event sourcing tests onto standard harnesses #5099

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
18 changes: 4 additions & 14 deletions src/EventSourcingTests/Bugs/Bug_4619_archive_stream_explicit_column_list_after_column_add.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,23 +36,13 @@ namespace EventSourcingTests.Bugs;
/// is robust to physical-order drift.
/// </para>
/// </summary>
public class Bug_4619_archive_stream_explicit_column_list_after_column_add
public class Bug_4619_archive_stream_explicit_column_list_after_column_add: BugIntegrationContext
{
[Fact]
public async Task archive_stream_succeeds_after_a_post_creation_column_add()
{
var schema = $"bug4619_{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 { }
}

using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = schema;
opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps;
// UseArchivedStreamPartitioning routes archives through the bulk
// function variant (writeWithPartitioning) — the broken positional
Expand Down Expand Up @@ -85,8 +75,8 @@ public async Task archive_stream_succeeds_after_a_post_creation_column_add()
await using (var conn = new NpgsqlConnection(ConnectionSource.ConnectionString))
{
await conn.OpenAsync();
await conn.CreateCommand($"alter table {schema}.mt_events drop column bdata").ExecuteNonQueryAsync();
await conn.CreateCommand($"alter table {schema}.mt_events add column bdata bytea null").ExecuteNonQueryAsync();
await conn.CreateCommand($"alter table {SchemaName}.mt_events drop column bdata").ExecuteNonQueryAsync();
await conn.CreateCommand($"alter table {SchemaName}.mt_events add column bdata bytea null").ExecuteNonQueryAsync();
}

// Archive — the positional INSERT inside mt_archive_stream would
Expand Down
26 changes: 4 additions & 22 deletions src/EventSourcingTests/Bugs/Bug_4625_bulk_insert_events_derives_aggregate_type_name.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,22 +33,13 @@ namespace EventSourcingTests.Bugs;
/// callers needing to reach for the <c>internal set</c>.
/// </para>
/// </summary>
public class Bug_4625_bulk_insert_events_derives_aggregate_type_name
public class Bug_4625_bulk_insert_events_derives_aggregate_type_name: BugIntegrationContext
{
[Fact]
public async Task BulkInsertEventsAsync_writes_mt_streams_type_from_AggregateType()
{
var schema = $"bug4625_{Environment.ProcessId}_{Guid.NewGuid():N}".Substring(0, 32);
await using (var conn = new NpgsqlConnection(ConnectionSource.ConnectionString))
var store = StoreOptions(opts =>
{
await conn.OpenAsync();
try { await conn.DropSchemaAsync(schema); } catch { }
}

using var store = DocumentStore.For(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = schema;
opts.Events.TenancyStyle = TenancyStyle.Conjoined;
opts.Events.StreamIdentity = StreamIdentity.AsString;
opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps;
Expand All @@ -71,7 +62,7 @@ public async Task BulkInsertEventsAsync_writes_mt_streams_type_from_AggregateTyp
await using var conn2 = new NpgsqlConnection(ConnectionSource.ConnectionString);
await conn2.OpenAsync();
await using var cmd = conn2.CreateCommand(
$"select type from {schema}.mt_streams where id = @id and tenant_id = @tid");
$"select type from {SchemaName}.mt_streams where id = @id and tenant_id = @tid");
cmd.Parameters.AddWithValue("id", streamKey);
cmd.Parameters.AddWithValue("tid", "alpha");
var typeName = (string?)await cmd.ExecuteScalarAsync();
Expand All @@ -91,17 +82,8 @@ public async Task bulk_inserted_stream_with_AggregateType_supports_subsequent_Ap
// type was lost (NULL) can't take any further appends — the mandatory
// guard fires on every subsequent operation. Post-fix, the bulk-inserted
// stream's type IS set, so appends work.
var schema = $"bug4625b_{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 { }
}

using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = schema;
opts.Events.TenancyStyle = TenancyStyle.Conjoined;
opts.Events.StreamIdentity = StreamIdentity.AsString;
opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps;
Expand Down
44 changes: 8 additions & 36 deletions ...Projections/MultiStreamProjections/CustomGroupers/Bug_4261_multistream_sample_coverage.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ namespace EventSourcingTests.Projections.MultiStreamProjections.CustomGroupers;
/// https://github.com/JasperFx/marten/discussions/3615.
/// Pattern 3 should pass because the derived event carries the group key directly.
/// </summary>
public partial class Bug_4261_multistream_sample_coverage
public partial class Bug_4261_multistream_sample_coverage: OneOffConfigurationsContext
{
private readonly ITestOutputHelper _output;

Expand All @@ -44,10 +44,8 @@ public Bug_4261_multistream_sample_coverage(ITestOutputHelper output)
[Fact]
public async Task pattern1_async_same_batch_link_and_usage_is_applied_correctly()
{
await using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = $"b4261_p1_net{Environment.Version.Major}";
opts.Events.StreamIdentity = StreamIdentity.AsString;
opts.Events.AddEventType(typeof(P1.CustomerRegistered));
opts.Events.AddEventType(typeof(P1.CustomerLinkedToExternalAccount));
Expand All @@ -57,8 +55,6 @@ public async Task pattern1_async_same_batch_link_and_usage_is_applied_correctly(
opts.Projections.Add<P1.CustomerBillingProjection>(ProjectionLifecycle.Async);
});

await store.Advanced.Clean.CompletelyRemoveAllAsync();

using var daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();

Expand Down Expand Up @@ -93,10 +89,8 @@ public async Task pattern1_async_usage_before_link_in_separate_batches_works()
{
// Sanity baseline: when the link is committed first and the usage arrives
// in a later batch, Pattern 1 should Just Work.
await using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = $"b4261_p1_seq_net{Environment.Version.Major}";
opts.Events.StreamIdentity = StreamIdentity.AsString;
opts.Events.AddEventType(typeof(P1.CustomerRegistered));
opts.Events.AddEventType(typeof(P1.CustomerLinkedToExternalAccount));
Expand All @@ -106,8 +100,6 @@ public async Task pattern1_async_usage_before_link_in_separate_batches_works()
opts.Projections.Add<P1.CustomerBillingProjection>(ProjectionLifecycle.Async);
});

await store.Advanced.Clean.CompletelyRemoveAllAsync();

using var daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();

Expand Down Expand Up @@ -157,10 +149,8 @@ public async Task pattern2_async_same_batch_loses_usage_event_known_limitation()
//
// If a future engine change makes Pattern 2 work under same-batch ordering,
// this test will fail and should be retired.
await using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = $"b4261_p2_net{Environment.Version.Major}";
opts.Events.StreamIdentity = StreamIdentity.AsString;
opts.Events.AddEventType(typeof(P2.CustomerRegistered));
opts.Events.AddEventType(typeof(P2.CustomerLinkedToExternalAccount));
Expand All @@ -169,8 +159,6 @@ public async Task pattern2_async_same_batch_loses_usage_event_known_limitation()
opts.Projections.Add<P2.CustomerBillingProjection>(ProjectionLifecycle.Async);
});

await store.Advanced.Clean.CompletelyRemoveAllAsync();

using var daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();

Expand Down Expand Up @@ -204,10 +192,8 @@ public async Task pattern2_async_link_in_earlier_batch_then_usage_works()
{
// Sanity baseline: when the link arrives first and is already applied to the
// projection, Pattern 2's containment query in a later batch finds the owner.
await using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = $"b4261_p2_seq_net{Environment.Version.Major}";
opts.Events.StreamIdentity = StreamIdentity.AsString;
opts.Events.AddEventType(typeof(P2.CustomerRegistered));
opts.Events.AddEventType(typeof(P2.CustomerLinkedToExternalAccount));
Expand All @@ -216,8 +202,6 @@ public async Task pattern2_async_link_in_earlier_batch_then_usage_works()
opts.Projections.Add<P2.CustomerBillingProjection>(ProjectionLifecycle.Async);
});

await store.Advanced.Clean.CompletelyRemoveAllAsync();

using var daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();

Expand Down Expand Up @@ -259,10 +243,8 @@ public async Task pattern4_async_same_batch_link_and_usage_works()
// to pick up link events that share the same daemon cycle as the usage
// event. This is the recommended pattern when link+usage events can
// appear in a single SaveChangesAsync.
await using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = $"b4261_p4_net{Environment.Version.Major}";
opts.Events.StreamIdentity = StreamIdentity.AsString;
opts.Events.AddEventType(typeof(P4.CustomerRegistered));
opts.Events.AddEventType(typeof(P4.CustomerLinkedToExternalAccount));
Expand All @@ -272,8 +254,6 @@ public async Task pattern4_async_same_batch_link_and_usage_works()
opts.Projections.Add<P4.CustomerBillingProjection>(ProjectionLifecycle.Async);
});

await store.Advanced.Clean.CompletelyRemoveAllAsync();

using var daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();

Expand Down Expand Up @@ -304,10 +284,8 @@ public async Task pattern4_async_link_in_earlier_batch_then_usage_works()
{
// Baseline: Pattern 4 must also handle the case where the link event
// was committed in a prior batch. The DB fallback covers that.
await using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = $"b4261_p4_seq_net{Environment.Version.Major}";
opts.Events.StreamIdentity = StreamIdentity.AsString;
opts.Events.AddEventType(typeof(P4.CustomerRegistered));
opts.Events.AddEventType(typeof(P4.CustomerLinkedToExternalAccount));
Expand All @@ -317,8 +295,6 @@ public async Task pattern4_async_link_in_earlier_batch_then_usage_works()
opts.Projections.Add<P4.CustomerBillingProjection>(ProjectionLifecycle.Async);
});

await store.Advanced.Clean.CompletelyRemoveAllAsync();

using var daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();

Expand Down Expand Up @@ -357,17 +333,13 @@ public async Task pattern3_async_same_batch_is_correct_by_design()
{
// Pattern 3 keeps the grouping key (CustomerId) on the terminal event itself,
// so same-batch ordering cannot create a race.
await using var store = DocumentStore.For(opts =>
var store = StoreOptions(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = $"b4261_p3_net{Environment.Version.Major}";
opts.Events.AddEventType(typeof(P3.ShipmentBilled));

opts.Projections.Add<P3.CustomerBillingProjection>(ProjectionLifecycle.Async);
});

await store.Advanced.Clean.CompletelyRemoveAllAsync();

using var daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();

Expand Down
30 changes: 6 additions & 24 deletions src/EventSourcingTests/rebuild_concurrency_cap_resolution.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,29 +12,19 @@ namespace EventSourcingTests;

// jasperfx#420 / marten#4710: resolution of the per-database rebuild concurrency cap
// surfaced through IEventStore.MaxConcurrentRebuildsPerDatabase.
public class rebuild_concurrency_cap_resolution
public class rebuild_concurrency_cap_resolution: OneOffConfigurationsContext
{
private static DocumentStore storeWith(Action<StoreOptions> configure)
{
return DocumentStore.For(opts =>
{
opts.Connection(ConnectionSource.ConnectionString);
opts.DatabaseSchemaName = "rebuild_cap";
configure(opts);
});
}

[Fact]
public void configured_value_wins_over_derived_default()
{
using var store = storeWith(opts => opts.Projections.MaxConcurrentRebuildsPerDatabase = 3);
var store = SeparateStore(opts => opts.Projections.MaxConcurrentRebuildsPerDatabase = 3);
((IEventStore)store).MaxConcurrentRebuildsPerDatabase.ShouldBe(3);
}

[Fact]
public void non_positive_configured_value_disables_the_cap()
{
using var store = storeWith(opts => opts.Projections.MaxConcurrentRebuildsPerDatabase = 0);
var store = SeparateStore(opts => opts.Projections.MaxConcurrentRebuildsPerDatabase = 0);
((IEventStore)store).MaxConcurrentRebuildsPerDatabase.ShouldBeNull();
}

Expand All @@ -46,11 +36,7 @@ public void derived_default_is_pool_size_over_eight_with_floor_of_one()
MaxPoolSize = 64
}.ConnectionString;

using var store = DocumentStore.For(opts =>
{
opts.Connection(connectionString);
opts.DatabaseSchemaName = "rebuild_cap";
});
var store = SeparateStore(opts => opts.Connection(connectionString));

((IEventStore)store).MaxConcurrentRebuildsPerDatabase.ShouldBe(8);
}
Expand All @@ -63,11 +49,7 @@ public void derived_default_floors_at_one_for_tiny_pools()
MaxPoolSize = 5
}.ConnectionString;

using var store = DocumentStore.For(opts =>
{
opts.Connection(connectionString);
opts.DatabaseSchemaName = "rebuild_cap";
});
var store = SeparateStore(opts => opts.Connection(connectionString));

((IEventStore)store).MaxConcurrentRebuildsPerDatabase.ShouldBe(1);
}
Expand All @@ -77,7 +59,7 @@ public async Task usage_descriptor_carries_the_effective_cap()
{
// jasperfx#434: CritterWatch#309's rebuild dispatcher reads the effective cap
// off the EventStoreUsage descriptor rather than guessing.
using var store = storeWith(opts => opts.Projections.MaxConcurrentRebuildsPerDatabase = 6);
var store = SeparateStore(opts => opts.Projections.MaxConcurrentRebuildsPerDatabase = 6);

var usage = await ((IEventStore)store).TryCreateUsage(CancellationToken.None);

Expand Down
Loading