ICompletableSynchronizedStorageSession is completed after storing Outbox messages

[timbussmann]

Describe the bug

Description

When using the outbox enabled transactional session, ICompletableSynchronizedStorageSession.CompleteAsync is completed after storing the outgoing messages to the Outbox.

When using SQLPersistence with the ISqlStorageSession, the ISqlStorageSession's OnCompleteAsync callback is not able to store additional outgoing messages to the Outbox which might lead to messages that are sent within that callback to be dropped.

Expected behavior

Messages sent within ISqlStorageSession.OnCompleteAsync should be stored to the Outbox too. This is more aligned with Core's behavior in the incoming pipeline, where the session is completed in the LoadHandlersConnector while the outbox Store operations happens in the TransportReceiveToPhysicalMessageConnector afterwards.

Versions

Fixed in:

• 2.0.2
• 1.0.3

Steps to reproduce

This can be reproduced by using the following EF configuration with the RF + TxSession sample:

                c.AddScoped(b =>
                {
                    var session = b.GetRequiredService<ISqlStorageSession>();
                    var context = new MyDataContext(new DbContextOptionsBuilder<MyDataContext>()
                        .UseSqlServer(session.Connection)
                        .Options);
                    var txSession = b.GetService<ITransactionalSession>();

                    //Use the same underlying ADO.NET transaction
                    context.Database.UseTransaction(session.Transaction);

                    //Ensure context is flushed before the transaction is committed
                    session.OnSaveChanges(async (s, token) =>
                    {
                        await context.SaveChangesAsync(token);
 
                        await txSession.SendLocal(new DomainEvent());
                    });

                    return context;

Additional Information

Note that this bug is only relevant for users using the transactional session with SQL Persistence via the NServiceBus.Persistence.Sql.TransactionalSession package