diff --git a/docs/events/bulk-appending.md b/docs/events/bulk-appending.md index 977cc0a328..05732c51c9 100644 --- a/docs/events/bulk-appending.md +++ b/docs/events/bulk-appending.md @@ -25,6 +25,16 @@ The bulk append API: This approach is significantly faster than the normal append path because it avoids per-row function calls, version checking, and individual INSERT statements. +### Schema Application + +`BulkInsertEventsAsync` will apply any outstanding schema changes **at most once per database**, and only +for the database the events are actually being imported into. It is *not* re-applied on every call, so a +conversion tool that loops batches over a single store pays the schema introspection once rather than per +batch. When the store's `AutoCreateSchemaObjects` is `AutoCreate.None` — schema managed out of band — the +apply is skipped entirely, since it cannot do anything by contract and would only cost introspection. + +As always with `AutoCreate.None`, the event storage must exist before you import. + ## Basic Usage Build a list of `StreamAction` objects representing new event streams, then call `BulkInsertEventsAsync` diff --git a/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs b/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs new file mode 100644 index 0000000000..0771edf9e1 --- /dev/null +++ b/src/EventSourcingTests/Bugs/Bug_4946_bulk_insert_does_not_apply_schema_per_call.cs @@ -0,0 +1,149 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using JasperFx; +using JasperFx.Events; +using Marten; +using Marten.Events; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace EventSourcingTests.Bugs; + +/// +/// GH-4946: the batch BulkInsertEventsAsync overloads ran +/// Storage.ApplyAllConfiguredChangesToDatabaseAsync() on *every* call, so a conversion tool +/// importing in 1,000-event batches paid a full schema delta — partition introspection plus +/// information_schema sweeps, across every database in the store — per batch. The regression is +/// *how many times* the apply runs, so that is what these specs assert. +/// +public class Bug_4946_bulk_insert_does_not_apply_schema_per_call: OneOffConfigurationsContext +{ + private static List streams(EventGraph events, int count) + { + var actions = new List(); + for (var i = 0; i < count; i++) + { + actions.Add(StreamAction.Start(events, Guid.NewGuid(), new object[] + { + new QuestStarted { Name = $"Quest {i}" }, new QuestEnded { Name = $"Quest {i}" } + })); + } + + return actions; + } + + [Fact] + public async Task apply_the_schema_at_most_once_across_many_batches() + { + var store = StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsGuid); + + for (var i = 0; i < 5; i++) + { + await store.BulkInsertEventsAsync(streams(store.Events, 2)); + } + + store.BulkInsertSchemaApplicationCount.ShouldBe(1); + + var stats = await store.Advanced.FetchEventStoreStatistics(); + stats.EventCount.ShouldBe(20); + } + + [Fact] + public async Task apply_the_schema_at_most_once_across_many_batches_for_a_tenant() + { + var store = StoreOptions(opts => + { + opts.Events.StreamIdentity = StreamIdentity.AsGuid; + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + }); + + for (var i = 0; i < 5; i++) + { + await store.BulkInsertEventsAsync("tenant-one", streams(store.Events, 2)); + } + + // Same single database behind every tenant here, so exactly one apply + store.BulkInsertSchemaApplicationCount.ShouldBe(1); + } + + [Fact] + public async Task concurrent_batches_only_apply_the_schema_once() + { + var store = StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsGuid); + + var batches = Enumerable.Range(0, 10) + .Select(_ => store.BulkInsertEventsAsync(streams(store.Events, 2))) + .ToArray(); + + await Task.WhenAll(batches); + + store.BulkInsertSchemaApplicationCount.ShouldBe(1); + } + + [Fact] + public async Task one_batch_cancelling_does_not_fail_a_concurrent_sibling_batch() + { + // The per-database schema apply is memoized, so the FIRST caller to arrive owns the single + // in-flight task that every concurrent caller for the same database awaits. If that shared + // task ran under the first caller's CancellationToken, then cancelling one batch would fail + // every sibling batch that happened to be waiting on it — batches that were never cancelled. + // The apply therefore runs untethered, and each caller awaits it under its OWN token. + var store = StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsGuid); + + using var doomed = new CancellationTokenSource(); + + var cancelled = store.BulkInsertEventsAsync(streams(store.Events, 2), cancellation: doomed.Token); + var survivor = store.BulkInsertEventsAsync(streams(store.Events, 2)); + + await doomed.CancelAsync(); + + // Whatever becomes of the cancelled batch, the sibling must land on its own merits + try + { + await cancelled; + } + catch (OperationCanceledException) + { + // expected, and immaterial to the assertion below + } + + await Should.NotThrowAsync(async () => await survivor); + + store.BulkInsertSchemaApplicationCount.ShouldBe(1); + } + + [Fact] + public async Task no_schema_apply_at_all_when_auto_create_is_none() + { + // Get the schema in place with a "normal" store first + var creator = StoreOptions(opts => + { + opts.Events.StreamIdentity = StreamIdentity.AsGuid; + opts.Events.AddEventTypes([typeof(QuestStarted), typeof(QuestEnded)]); + }); + + await creator.Storage.ApplyAllConfiguredChangesToDatabaseAsync(); + + // Same schema, but the schema is managed out of band now + var store = SeparateStore(opts => + { + opts.Events.StreamIdentity = StreamIdentity.AsGuid; + opts.AutoCreateSchemaObjects = AutoCreate.None; + }); + + for (var i = 0; i < 3; i++) + { + await store.BulkInsertEventsAsync(streams(store.Events, 2)); + } + + // The apply is a no-op by contract under AutoCreate.None, so it should never be attempted + store.BulkInsertSchemaApplicationCount.ShouldBe(0); + + var stats = await store.Advanced.FetchEventStoreStatistics(); + stats.EventCount.ShouldBe(12); + } +} diff --git a/src/Marten/DocumentStore.cs b/src/Marten/DocumentStore.cs index 42bfd17c4d..ab07a70191 100644 --- a/src/Marten/DocumentStore.cs +++ b/src/Marten/DocumentStore.cs @@ -1,5 +1,6 @@ #nullable enable using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Linq; @@ -189,12 +190,84 @@ public async Task BulkInsertDocumentsAsync(string tenantId, IEnumerable await bulkInsertion.BulkInsertDocumentsAsync(documents, mode, batchSize, cancellation).ConfigureAwait(false); } + /// + /// Tracks which databases have already had the schema apply run on behalf of the bulk event + /// insert path, keyed by database identifier. See . + /// + private readonly ConcurrentDictionary> _bulkInsertSchemaApplications = new(); + + /// + /// Test hook (GH-4946): how many times the bulk event insert path has actually executed the + /// schema apply. The regression being guarded is *how many times* the apply runs, not whether + /// the insert works. + /// + internal int BulkInsertSchemaApplicationCount; + + /// + /// GH-4946: the batch BulkInsertEventsAsync overloads used to call + /// Storage.ApplyAllConfiguredChangesToDatabaseAsync() on *every* call — a full schema delta + /// (partition introspection + information_schema sweeps) across *every* database in the store, per + /// batch. A caller importing in 1,000-event batches paid that per batch, which collapsed import + /// throughput and parked the connection pool on Weasel's partition introspection query. + /// + /// The apply is now (a) skipped altogether when the effective is + /// , where it is a no-op by contract and only the introspection cost + /// remains, and (b) otherwise run at most once per *database* — not per call, and not across + /// databases the import never touches. A 512-database sharded store therefore applies at most once + /// per database it actually imports into. + /// + /// + /// Thread safety: batches may be imported concurrently. The per-database entry is a + /// with , so + /// concurrent callers for the same database all await the single in-flight apply. A failed apply is + /// evicted so a later call can retry rather than caching the failure forever. + /// + /// + private Task ensureBulkInsertSchemaAsync(IMartenDatabase database, CancellationToken cancellation) + { + if (Options.AutoCreateSchemaObjects == AutoCreate.None) + { + return Task.CompletedTask; + } + + var lazy = _bulkInsertSchemaApplications.GetOrAdd(database.Identifier, + _ => new Lazy(() => applyAsync(database), + LazyThreadSafetyMode.ExecutionAndPublication)); + + // Await the shared apply under *this* caller's token. The apply itself deliberately does not + // take a caller's token: the first caller to arrive owns the in-flight task that every + // concurrent caller for the same database awaits, so honoring its token here would let one + // batch's cancellation fail a sibling batch that was never cancelled. A schema apply is short + // and idempotent, so letting it run to completion is the cheaper trade. + return lazy.Value.WaitAsync(cancellation); + + async Task applyAsync(IMartenDatabase db) + { + try + { + Interlocked.Increment(ref BulkInsertSchemaApplicationCount); + + // MA0040 suppressed deliberately: this task is SHARED by every concurrent caller for + // this database, so it must not be bound to any one caller's token. Each caller applies + // its own token at the await site above (`lazy.Value.WaitAsync(cancellation)`). +#pragma warning disable MA0040 + await db.ApplyAllConfiguredChangesToDatabaseAsync().ConfigureAwait(false); +#pragma warning restore MA0040 + } + catch + { + _bulkInsertSchemaApplications.TryRemove(db.Identifier, out _); + throw; + } + } + } + public async Task BulkInsertEventsAsync(IReadOnlyList streams, int batchSize = 1000, CancellationToken cancellation = default) { - await Storage.ApplyAllConfiguredChangesToDatabaseAsync().ConfigureAwait(false); - var tenant = Tenancy.Default; + await ensureBulkInsertSchemaAsync(tenant.Database, cancellation).ConfigureAwait(false); + var appender = new BulkEventAppender(Events, Options.Serializer()); await using var conn = tenant.Database.CreateConnection(); @@ -216,10 +289,11 @@ public async Task BulkInsertEventsAsync(IReadOnlyList streams, int public async Task BulkInsertEventsAsync(string tenantId, IReadOnlyList streams, int batchSize = 1000, CancellationToken cancellation = default) { - await Storage.ApplyAllConfiguredChangesToDatabaseAsync().ConfigureAwait(false); - var tenant = await Tenancy.GetTenantAsync(Options.TenantIdStyle.MaybeCorrectTenantId(tenantId)) .ConfigureAwait(false); + + await ensureBulkInsertSchemaAsync(tenant.Database, cancellation).ConfigureAwait(false); + var appender = new BulkEventAppender(Events, Options.Serializer()); // Set tenant on all streams