diff --git a/src/Http/Wolverine.Http.Tests/Bug_3291_lightweight_http_cascade_flushes_before_commit.cs b/src/Http/Wolverine.Http.Tests/Bug_3291_lightweight_http_cascade_flushes_before_commit.cs new file mode 100644 index 000000000..ea3dfedd0 --- /dev/null +++ b/src/Http/Wolverine.Http.Tests/Bug_3291_lightweight_http_cascade_flushes_before_commit.cs @@ -0,0 +1,138 @@ +using Alba; +using IntegrationTests; +using JasperFx; +using JasperFx.CodeGeneration; +using JasperFx.Core.Reflection; +using Marten; +using Microsoft.AspNetCore.Builder; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; +using Wolverine.EntityFrameworkCore; +using Wolverine.Marten; +using Wolverine.Persistence; +using WolverineWebApi; +using Xunit; + +namespace Wolverine.Http.Tests; + +// Reproducer for GH-3291. A Wolverine.Http endpoint has no incoming envelope, so (unlike a message +// handler, whose MessageContext is enlisted in the outbox by ReadEnvelope at runtime) its +// MessageContext.Transaction stays null. In TransactionMiddlewareMode.Lightweight the EF Core +// transaction middleware does NOT enroll the endpoint's DbContext/context in the outbox, so +// MessageBus.PersistOrSendAsync takes the StoreAndForwardAsync() (send-now) branch and the cascaded +// message is sent BEFORE the SaveChangesAsync postprocessor commits - silently dropping the +// transactional-outbox guarantee the HTTP docs advertise. Message handlers are unaffected in both +// modes; the bug is specific to HTTP endpoints in Lightweight mode. +// +// Like the sibling Eager-mode reproducer (Bug_efcore_outbox_flush_before_commit), we assert at the +// codegen surface rather than at runtime: the runtime symptom (a stranded wolverine_outgoing row) is +// cleaned up by the durability agent within ~250ms and races any post-request query. The generated +// composition is the deterministic proof. +// +// Pre-fix state (the bug): the Lightweight HTTP chain carries a standalone FlushOutgoingMessages +// postprocessor as its ONLY flush trigger, and its MessageContext is never enlisted, so the generated +// code never calls EnlistInOutboxAsync. Fixed state: the chain enlists the DbContext in the outbox +// WITHOUT an explicit BeginTransactionAsync (an IFlushesMessages middleware -> no standalone +// FlushOutgoingMessages), and the buffered messages are flushed after the commit. +public class Bug_3291_lightweight_http_cascade_flushes_before_commit +{ + private static async Task buildLightweightHostAsync() + { + var schema = "ef_lw_" + Guid.NewGuid().ToString("N")[..8]; + var builder = WebApplication.CreateBuilder(); + + // Wolverine-integrated DbContext supplies the outbox enrollment for EF Core (mirrors the + // AddDbContextWithWolverineIntegration setup the issue reports). + builder.Services.AddDbContextWithWolverineIntegration(x => + x.UseNpgsql(Servers.PostgresConnectionString)); + + builder.Host.UseWolverine(opts => + { + opts.Durability.Mode = DurabilityMode.Solo; + + opts.Services.AddMarten(m => + { + m.Connection(Servers.PostgresConnectionString); + m.DatabaseSchemaName = schema; + }).IntegrateWithWolverine(); + + // The bug is specific to Lightweight mode. Eager already works (see the sibling reproducer). + opts.UseEntityFrameworkCoreTransactions(TransactionMiddlewareMode.Lightweight); + + opts.Policies.AutoApplyTransactions(); + opts.Policies.UseDurableLocalQueues(); + + opts.Discovery.DisableConventionalDiscovery(); + // The cascaded ItemCreated needs a routed handler so it actually buffers into Outstanding. + opts.Discovery.IncludeType(); + opts.Discovery.IncludeAssembly(typeof(Bug_3291_lightweight_http_cascade_flushes_before_commit).Assembly); + }); + + builder.Services.AddWolverineHttp(); + + return await AlbaHost.For(builder, app => app.MapWolverineEndpoints()); + } + + [Fact] + public async Task lightweight_http_endpoint_enlists_outbox_and_does_not_flush_before_commit() + { + await using var host = await buildLightweightHostAsync(); + + var graph = host.Services.GetRequiredService().Endpoints!; + var chain = graph.ChainFor("POST", "/ef/lightweight/publish"); + chain.ShouldNotBeNull(); + + // (1) No standalone pre-commit flush. Pre-fix, applyEagerCommitOrLightweightFlush adds a + // FlushOutgoingMessages postprocessor that (because the context is never enlisted) runs the + // send BEFORE SaveChangesAsync commits. Fixed, the enlist middleware is IFlushesMessages, so + // this standalone flush is gone and the commit path does the post-commit flush instead. + chain.Postprocessors.OfType().ShouldBeEmpty( + "GH-3291: a Lightweight-mode HTTP endpoint that cascades messages still has a standalone " + + "FlushOutgoingMessages postprocessor. Its MessageContext is never enlisted in the outbox, " + + "so the cascade is sent before SaveChangesAsync commits."); + + // (2) The generated code must enlist the DbContext + IMessageContext in the outbox so the + // cascade buffers and flushes after commit. Pre-fix this call is absent. + // GH-3291: pre-fix the Lightweight HTTP endpoint never enrolls its DbContext/context in the + // outbox, so cascaded messages are sent immediately instead of buffered until after the commit. + chain.As().InitializeSynchronously(graph.Rules, graph, host.Services); + var source = chain!.SourceCode; + source.ShouldNotBeNull(); + source.ShouldContain("EnlistInOutboxAsync"); + + // (3) Ordering: enroll in the outbox -> [endpoint body] -> SaveChangesAsync commits -> the + // envelope transaction's CommitAsync flushes the buffered messages. The flush must come AFTER + // the commit; that is the whole point of the fix. + // Match the actual call sites (".Method(") rather than bare names, which also appear in comments. + var enlistAt = source.IndexOf(".EnlistInOutboxAsync(", StringComparison.Ordinal); + var saveAt = source.IndexOf(".SaveChangesAsync(", StringComparison.Ordinal); + var commitAt = source.IndexOf(".CommitAsync(", StringComparison.Ordinal); + + saveAt.ShouldBeGreaterThan(enlistAt, "SaveChangesAsync must run after the outbox enrollment"); + commitAt.ShouldBeGreaterThan(saveAt, + "The outbox flush (EfCoreEnvelopeTransaction.CommitAsync) must run after SaveChangesAsync commits"); + } +} + +public static class LightweightEfCascadeEndpoint +{ + // Writes an entity AND cascades a message: the exact shape from the GH-3291 report. In Lightweight + // mode the cascade must not be sent until SaveChangesAsync commits the item. + [WolverinePost("/ef/lightweight/publish")] + public static async Task Publish(CreateItemCommand command, ItemsDbContext db, IMessageBus bus) + { + var item = new Item { Name = command.Name }; + db.Items.Add(item); + await bus.PublishAsync(new ItemCreated { Id = item.Id }); + } +} + +// A routed handler so the cascaded ItemCreated has somewhere to go (durable local queue). Body is +// irrelevant — the test asserts the outbox enrollment/flush composition, not the handling. +public class LightweightCascadeItemCreatedHandler +{ + public void Handle(ItemCreated _) + { + } +} diff --git a/src/Persistence/Wolverine.EntityFrameworkCore/Codegen/EFCorePersistenceFrameProvider.cs b/src/Persistence/Wolverine.EntityFrameworkCore/Codegen/EFCorePersistenceFrameProvider.cs index cd7870c32..040e34ecb 100644 --- a/src/Persistence/Wolverine.EntityFrameworkCore/Codegen/EFCorePersistenceFrameProvider.cs +++ b/src/Persistence/Wolverine.EntityFrameworkCore/Codegen/EFCorePersistenceFrameProvider.cs @@ -226,6 +226,24 @@ public void ApplyTransactionSupport(IChain chain, IServiceContainer container) enrolledInTransaction = true; } } + else if (isHttpChain(chain) + && !isMultiTenanted(container, dbContextType) + && chain.RequiresOutbox() + && chain.ShouldFlushOutgoingMessages()) + { + // GH-3291: A Wolverine.Http endpoint has no incoming envelope, so - unlike a message handler, + // whose MessageContext is enlisted by MessageContext.ReadEnvelope at runtime - its + // MessageContext.Transaction stays null in Lightweight mode. Cascaded messages would then be + // sent immediately, before the SaveChangesAsync postprocessor commits, silently dropping the + // transactional-outbox guarantee for HTTP endpoints (message handlers are unaffected). Enroll + // the DbContext in the outbox WITHOUT an explicit BeginTransactionAsync (SaveChanges' implicit + // transaction covers the write, and skipping the explicit begin keeps this compatible with EF + // Core's EnableRetryOnFailure). Setting enrolledInTransaction = true makes the code below add + // the CommitEfCoreEnvelopeTransaction postprocessor (which flushes after commit) instead of a + // standalone, pre-commit FlushOutgoingMessages. Restricted to HttpChain on purpose. + chain.Middleware.Insert(0, new EnlistDbContextInOutbox(dbContextType)); + enrolledInTransaction = true; + } var abstractionType = chain.ServiceDependencies(container, Type.EmptyTypes).FirstOrDefault(x => _abstractions.Contains(x)); if (abstractionType != null) @@ -323,6 +341,13 @@ private bool isMultiTenanted(IServiceContainer container, Type dbContextType) return container.HasRegistrationFor(typeof(IDbContextBuilder<>).MakeGenericType(dbContextType)); } + // HttpChain is the only chain type whose Scoping is HttpEndpoints. Detected via the scoping enum + // rather than a type reference so Wolverine.EntityFrameworkCore need not depend on Wolverine.Http. + private static bool isHttpChain(IChain chain) + { + return chain.Scoping == MiddlewareScoping.HttpEndpoints; + } + public void ApplyTransactionSupport(IChain chain, IServiceContainer container, Type entityType) { // GH-3039: For saga chains, defer to the saga's own transaction-support application at codegen diff --git a/src/Persistence/Wolverine.EntityFrameworkCore/Codegen/EnlistDbContextInOutbox.cs b/src/Persistence/Wolverine.EntityFrameworkCore/Codegen/EnlistDbContextInOutbox.cs new file mode 100644 index 000000000..cd100d42a --- /dev/null +++ b/src/Persistence/Wolverine.EntityFrameworkCore/Codegen/EnlistDbContextInOutbox.cs @@ -0,0 +1,83 @@ +using JasperFx.CodeGeneration; +using JasperFx.CodeGeneration.Frames; +using JasperFx.CodeGeneration.Model; +using JasperFx.Core.Reflection; +using Wolverine.EntityFrameworkCore.Internals; +using Wolverine.Persistence; +using Wolverine.Runtime; + +namespace Wolverine.EntityFrameworkCore.Codegen; + +/// +/// GH-3291: enrolls a Wolverine-enabled DbContext + the IMessageContext in the outgoing outbox +/// transaction WITHOUT beginning an explicit database transaction. Used for Wolverine.Http endpoints in +/// mode. +/// +/// Unlike a message handler — whose MessageContext is enlisted at runtime by +/// MessageContext.ReadEnvelope when it reads the incoming envelope — an HTTP endpoint has no +/// incoming envelope, so its MessageContext.Transaction is otherwise null. In Lightweight mode +/// that means MessageBus.PersistOrSendAsync takes the send-now branch and cascaded messages are +/// dispatched BEFORE the SaveChangesAsync postprocessor commits. Enlisting here makes those +/// cascades buffer and flush after the commit instead. +/// +/// This deliberately does NOT call BeginTransactionAsync (that is what +/// does for Eager mode): the write is covered by the implicit +/// transaction SaveChangesAsync opens, and skipping the explicit begin keeps this compatible with +/// EF Core's retrying execution strategy (EnableRetryOnFailure), which forbids user-initiated +/// transactions. Implements so HttpChain does not also add a +/// standalone (pre-commit) FlushOutgoingMessages; the paired +/// postprocessor performs the post-commit flush. +/// +internal class EnlistDbContextInOutbox : AsyncFrame, IFlushesMessages +{ + private readonly Type _dbContextType; + private readonly Variable _envelopeTransaction; + private Variable _dbContext = null!; + private Variable? _context; + private Variable _scrapers = null!; + + public EnlistDbContextInOutbox(Type dbContextType) + { + _dbContextType = dbContextType; + _envelopeTransaction = new Variable(typeof(EfCoreEnvelopeTransaction), this); + } + + public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) + { + writer.WriteLine(""); + writer.WriteComment( + "GH-3291: enroll the DbContext & IMessagingContext in the outbox so cascaded messages buffer"); + writer.WriteComment( + "and flush AFTER SaveChangesAsync commits. No explicit transaction is started (Lightweight mode)."); + writer.Write($"var {_envelopeTransaction.Usage} = new {typeof(EfCoreEnvelopeTransaction).FullNameInCode()}({_dbContext.Usage}, {_context!.Usage}, {_scrapers.Usage});"); + writer.Write($"await {_context.Usage}.{nameof(MessageContext.EnlistInOutboxAsync)}({_envelopeTransaction.Usage}).ConfigureAwait(false);"); + + Next?.GenerateCode(method, writer); + } + + public override void GenerateFSharpCode(GeneratedMethod method, ISourceWriter writer) + { + // Mirrors the C# body inside an async `task { }` computation expression: awaits become `do!` + // and `.ConfigureAwait(false)` is dropped (the CE controls scheduling). See + // EnrollDbContextInTransaction for the same conventions. + writer.Write(""); + writer.WriteComment( + "GH-3291: enroll the DbContext & IMessagingContext in the outbox (Lightweight mode, no explicit transaction)"); + writer.Write($"{_envelopeTransaction.FSharpAssignmentUsage} = {typeof(EfCoreEnvelopeTransaction).FSharpName()}({_dbContext.FSharpUsage}, {_context!.FSharpUsage}, {_scrapers.FSharpUsage})"); + writer.Write($"do! {_context.FSharpUsage}.{nameof(MessageContext.EnlistInOutboxAsync)}({_envelopeTransaction.FSharpUsage})"); + + Next?.GenerateFSharpCode(method, writer); + } + + public override IEnumerable FindVariables(IMethodVariables chain) + { + _scrapers = chain.FindVariable(typeof(IEnumerable)); + yield return _scrapers; + + _context = chain.FindVariable(typeof(MessageContext)); + yield return _context; + + _dbContext = chain.FindVariable(_dbContextType); + yield return _dbContext; + } +}