Don't dispose a DbDataSource the message store does not own

src/Persistence/EfCoreTests.MultiTenancy/MultiTenancyCompliance.cs

@@ -34,6 +34,19 @@ public abstract class MultiTenancyCompliance : IAsyncLifetime, IWolverineExtensi
     protected string tenant1ConnectionString = null!;
     protected string tenant2ConnectionString = null!;
     protected string tenant3ConnectionString = null!;
+
+    // Data sources a test hands to Wolverine. Wolverine does not dispose a caller's data source, and
+    // NpgsqlConnection.ClearAllPools() never reaches a data source's own pool, so the test disposes them
+    private readonly List<IAsyncDisposable> _ownedDataSources = [];
+
+    /// <summary>
+    /// Register a data source this test creates and gives to Wolverine, so it is disposed with the host
+    /// </summary>
+    protected T OwnedByTheTest<T>(T dataSource) where T : IAsyncDisposable
+    {
+        _ownedDataSources.Add(dataSource);
+        return dataSource;
+    }
 
     protected MultiTenancyCompliance(DatabaseEngine engine)
     {
@@ -102,6 +115,20 @@ public async ValueTask DisposeAsync()
             // Swallow errors from inner WebApplicationFactory dispose
         }
 
+        foreach (var dataSource in _ownedDataSources)
+        {
+            try
+            {
+                await dataSource.DisposeAsync();
+            }
+            catch (Exception)
+            {
+                // Swallow errors disposing a data source the host may already have closed
+            }
+        }
+
+        _ownedDataSources.Clear();
+
         NpgsqlConnection.ClearAllPools();
         SqlConnection.ClearAllPools();
     }

src/Persistence/EfCoreTests.MultiTenancy/multi_tenancy_with_static_tenants_and_data_sources_for_postgresql.cs

@@ -27,12 +27,13 @@ public multi_tenancy_with_static_tenants_and_data_sources_for_postgresql() : bas
 
     public override void Configure(WolverineOptions opts)
     {
-        opts.PersistMessagesWithPostgresql(NpgsqlDataSource.Create(Servers.PostgresConnectionString), "static_multi_tenancy")
+        // Wolverine does not dispose a data source it is given, so the test owns and disposes these
+        opts.PersistMessagesWithPostgresql(OwnedByTheTest(NpgsqlDataSource.Create(Servers.PostgresConnectionString)), "static_multi_tenancy")
             .RegisterStaticTenantsByDataSource(tenants =>
             {
-                tenants.Register("red", NpgsqlDataSource.Create(tenant1ConnectionString));
-                tenants.Register("blue", NpgsqlDataSource.Create(tenant2ConnectionString));
-                tenants.Register("green", NpgsqlDataSource.Create(tenant3ConnectionString));
+                tenants.Register("red", OwnedByTheTest(NpgsqlDataSource.Create(tenant1ConnectionString)));
+                tenants.Register("blue", OwnedByTheTest(NpgsqlDataSource.Create(tenant2ConnectionString)));
+                tenants.Register("green", OwnedByTheTest(NpgsqlDataSource.Create(tenant3ConnectionString)));
             });
 
         // Little weird, but we have to remove this DbContext to use

src/Persistence/MartenTests/Bugs/Bug_message_store_disposes_martens_data_source.cs

@@ -0,0 +1,84 @@
+using IntegrationTests;
+using JasperFx.Resources;
+using Marten;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Npgsql;
+using Shouldly;
+using Wolverine;
+using Wolverine.Marten;
+using Wolverine.Persistence.Durability;
+
+namespace MartenTests.Bugs;
+
+/// <summary>
+/// The Wolverine message store built by IntegrateWithWolverine() sits on Marten's own NpgsqlDataSource, and
+/// MessageDatabase.DisposeAsync disposed it unconditionally. Disposing an NpgsqlDataSource aborts every physical
+/// connection rented from it, so Wolverine's teardown broke a DocumentStore that was still in use, and disposed a
+/// data source the application gave Marten through UseNpgsqlDataSource(), which Marten itself stopped disposing in
+/// Marten #4874.
+/// </summary>
+public class Bug_message_store_disposes_martens_data_source : PostgresqlContext
+{
+    private const string SchemaName = "marten_ds_ownership";
+
+    [Fact]
+    public async Task stopping_the_host_does_not_dispose_a_data_source_given_to_marten()
+    {
+        await using var dataSource = NpgsqlDataSource.Create(Servers.PostgresConnectionString);
+
+        var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Durability.Mode = DurabilityMode.Solo;
+
+                opts.Services.AddSingleton(dataSource);
+                opts.Services.AddMarten(m =>
+                {
+                    m.DatabaseSchemaName = SchemaName;
+                    m.DisableNpgsqlLogging = true;
+                }).UseNpgsqlDataSource().IntegrateWithWolverine();
+
+                opts.Services.AddResourceSetupOnStartup();
+            }).StartAsync(TestContext.Current.CancellationToken);
+
+        await host.StopAsync(TestContext.Current.CancellationToken);
+        host.Dispose();
+
+        await using var command = dataSource.CreateCommand("select 1");
+        (await command.ExecuteScalarAsync(TestContext.Current.CancellationToken)).ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task disposing_the_message_store_leaves_the_document_store_usable()
+    {
+        using var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Durability.Mode = DurabilityMode.Solo;
+
+                opts.Services.AddMarten(m =>
+                {
+                    m.Connection(Servers.PostgresConnectionString);
+                    m.DatabaseSchemaName = SchemaName;
+                    m.DisableNpgsqlLogging = true;
+                }).IntegrateWithWolverine();
+
+                opts.Services.AddResourceSetupOnStartup();
+            }).StartAsync(TestContext.Current.CancellationToken);
+
+        await host.Services.GetRequiredService<IMessageStore>().DisposeAsync();
+
+        // The data source is Marten's, and Marten is still running
+        await using var session = host.Services.GetRequiredService<IDocumentStore>().LightweightSession();
+        session.Store(new DataSourceOwnershipDoc { Id = Guid.NewGuid() });
+        await session.SaveChangesAsync(TestContext.Current.CancellationToken);
+
+        await host.StopAsync(TestContext.Current.CancellationToken);
+    }
+}
+
+public class DataSourceOwnershipDoc
+{
+    public Guid Id { get; set; }
+}

src/Persistence/MySql/MySqlTests/Bug_message_store_disposes_data_source_it_does_not_own.cs

@@ -0,0 +1,35 @@
+using IntegrationTests;
+using Microsoft.Extensions.Hosting;
+using MySqlConnector;
+using Shouldly;
+using Wolverine;
+using Wolverine.MySql;
+
+namespace MySqlTests;
+
+/// <summary>
+/// MessageDatabase.DisposeAsync disposed its data source unconditionally, including one the caller passed to
+/// PersistMessagesWithMySql(MySqlDataSource). Wolverine now disposes only a data source it built itself.
+/// </summary>
+[Collection("mysql")]
+public class Bug_message_store_disposes_data_source_it_does_not_own
+{
+    [Fact]
+    public async Task stopping_the_host_does_not_dispose_a_caller_supplied_data_source()
+    {
+        await using var dataSource = new MySqlDataSourceBuilder(Servers.MySqlConnectionString).Build();
+
+        var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.PersistMessagesWithMySql(dataSource, "ds_ownership");
+            }).StartAsync(TestContext.Current.CancellationToken);
+
+        await host.StopAsync(TestContext.Current.CancellationToken);
+        host.Dispose();
+
+        await using var command = dataSource.CreateCommand("select 1");
+        Convert.ToInt32(await command.ExecuteScalarAsync(TestContext.Current.CancellationToken)).ShouldBe(1);
+    }
+}

src/Persistence/MySql/Wolverine.MySql/MySqlBackedPersistence.cs

@@ -94,7 +94,8 @@ public interface IMySqlBackedPersistence
 
     /// <summary>
     /// Opt into using static per-tenant database multi-tenancy with MySqlDataSource. With this option, Wolverine is assuming that
-    /// there the number of tenant databases is static and does not change at runtime
+    /// there the number of tenant databases is static and does not change at runtime. The data sources are owned by
+    /// the caller, and Wolverine does not dispose them
     /// </summary>
     /// <param name="configure"></param>
     /// <returns></returns>
@@ -110,7 +111,7 @@ public interface IMySqlBackedPersistence
 
     /// <summary>
     /// Opt into multi-tenancy with separate databases using your own strategy for finding the right connection string
-    /// for a given tenant id
+    /// for a given tenant id. The data sources it returns are owned by the caller, and Wolverine does not dispose them
     /// </summary>
     /// <param name="tenantSource"></param>
     /// <returns></returns>
@@ -225,13 +226,16 @@ public IMessageStore BuildMessageStore(IWolverineRuntime runtime)
 
         var sagaTables = runtime.Services.GetServices<SagaTableDefinition>().ToArray();
 
+        // Only a data source built here from the connection string is Wolverine's to dispose. One passed to
+        // PersistMessagesWithMySql(MySqlDataSource) belongs to the caller
+        var ownsMainSource = DataSource == null;
         var mainSource = DataSource ?? MySqlDataSourceFactory.Create(ConnectionString!);
         var logger = runtime.LoggerFactory.CreateLogger<MySqlMessageStore>();
 
         if (UseMasterTableTenancy)
         {
             var defaultStore = new MySqlMessageStore(settings, runtime.DurabilitySettings, mainSource,
-                logger, sagaTables);
+                logger, sagaTables, ownsMainSource);
 
             ConnectionStringTenancy = new MasterTenantSource(defaultStore, runtime.Options);
 
@@ -242,14 +246,14 @@ public IMessageStore BuildMessageStore(IWolverineRuntime runtime)
         if (ConnectionStringTenancy != null || DataSourceTenancy != null)
         {
             var defaultStore = new MySqlMessageStore(settings, runtime.DurabilitySettings, mainSource,
-                logger, sagaTables);
+                logger, sagaTables, ownsMainSource);
 
             return new MultiTenantedMessageStore(defaultStore, runtime,
                 new MySqlTenantedMessageStore(runtime, this, sagaTables));
         }
 
         return new MySqlMessageStore(settings, runtime.DurabilitySettings, mainSource,
-            logger, sagaTables);
+            logger, sagaTables, ownsMainSource);
     }
 
     public MessageStoreRole Role { get; set; } = MessageStoreRole.Main;

src/Persistence/MySql/Wolverine.MySql/MySqlConfigurationExtensions.cs

@@ -56,7 +56,7 @@ public static IMySqlBackedPersistence PersistMessagesWithMySql(this WolverineOpt
     ///     Register MySQL backed message persistence to a known MySqlDataSource
     /// </summary>
     /// <param name="options"></param>
-    /// <param name="dataSource"></param>
+    /// <param name="dataSource">Owned by the caller. Wolverine does not dispose it, so it can be shared with the rest of the application</param>
     /// <param name="schemaName">Optional schema name for the Wolverine envelope storage. This has to be a single database identifier; a multi-part name like "crm.sales" is rejected. See GH-3997</param>
     /// <param name="role">Default is Main. Use this to mark some stores as Ancillary to disambiguate the main storage for Wolverine</param>
     public static IMySqlBackedPersistence PersistMessagesWithMySql(this WolverineOptions options,

src/Persistence/MySql/Wolverine.MySql/MySqlMessageStore.cs

@@ -79,9 +79,9 @@ public MySqlMessageStore(DatabaseSettings databaseSettings, DurabilitySettings s
         Justification = "DatabaseSagaSchema<,> closed over runtime saga / id types at startup; AOT consumers preserve via TrimmerRootDescriptor. See AOT guide / #2769.")]
     public MySqlMessageStore(DatabaseSettings databaseSettings, DurabilitySettings settings,
         MySqlDataSource dataSource,
-        ILogger<MySqlMessageStore> logger, IEnumerable<SagaTableDefinition> sagaTypes) : base(databaseSettings,
+        ILogger<MySqlMessageStore> logger, IEnumerable<SagaTableDefinition> sagaTypes, bool ownsDataSource = true) : base(databaseSettings,
         dataSource,
-        settings, logger, new MySqlMigrator(), MySqlProvider.Instance)
+        settings, logger, new MySqlMigrator(), MySqlProvider.Instance, ownsDataSource)
     {
         _findAtLargeEnvelopesSql =
             $"SELECT {DatabaseConstants.IncomingFields} FROM {SchemaName}.{DatabaseConstants.IncomingTable} WHERE owner_id = {TransportConstants.AnyNode} AND status = '{EnvelopeStatus.Incoming}' AND {DatabaseConstants.ReceivedAt} = @address LIMIT @limit";

src/Persistence/MySql/Wolverine.MySql/MySqlTenantedMessageStore.cs

@@ -112,8 +112,9 @@ private MySqlMessageStore buildTenantStoreForDataSource(MySqlDataSource source)
             SchemaName = schemaNameFor(source.ConnectionString)
         };
 
+        // The tenant's data source was registered by the caller, who owns it
         var store = new MySqlMessageStore(settings, _runtime.Options.Durability, source,
-            _runtime.LoggerFactory.CreateLogger<MySqlMessageStore>(), _sagaTables);
+            _runtime.LoggerFactory.CreateLogger<MySqlMessageStore>(), _sagaTables, ownsDataSource: false);
 
         // Same naming as the connection string path above. Left unset, every tenant registered by
         // MySqlDataSource kept the "default" name -- and both MultiTenantedQueueListener and

src/Persistence/PostgresqlTests/Bugs/Bug_message_store_disposes_data_source_it_does_not_own.cs

@@ -0,0 +1,107 @@
+using IntegrationTests;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Npgsql;
+using Shouldly;
+using Weasel.Postgresql;
+using Weasel.Postgresql.Migrations;
+using Wolverine;
+using Wolverine.Persistence.Durability;
+using Wolverine.Postgresql;
+using Wolverine.RDBMS;
+
+namespace PostgresqlTests.Bugs;
+
+/// <summary>
+/// MessageDatabase.DisposeAsync disposed its data source unconditionally, including one the caller passed to
+/// PersistMessagesWithPostgresql(NpgsqlDataSource) or registered as a tenant database. Disposing an
+/// NpgsqlDataSource aborts every physical connection rented from it, so stopping a host broke everything else
+/// still using that data source. Wolverine now disposes only a data source it built itself.
+/// </summary>
+public class Bug_message_store_disposes_data_source_it_does_not_own : PostgresqlContext
+{
+    private const string SchemaName = "ds_ownership";
+
+    [Fact]
+    public async Task stopping_the_host_does_not_dispose_a_caller_supplied_data_source()
+    {
+        await using var dataSource = NpgsqlDataSource.Create(Servers.PostgresConnectionString);
+
+        var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.PersistMessagesWithPostgresql(dataSource, SchemaName);
+            }).StartAsync(TestContext.Current.CancellationToken);
+
+        await host.StopAsync(TestContext.Current.CancellationToken);
+        host.Dispose();
+
+        await shouldStillBeUsable(dataSource);
+    }
+
+    [Fact]
+    public async Task stopping_the_host_does_not_dispose_a_caller_supplied_tenant_data_source()
+    {
+        var tenantConnectionString = await createDatabaseIfNotExists("ds_ownership_tenant");
+
+        await using var tenantDataSource = NpgsqlDataSource.Create(tenantConnectionString);
+
+        var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, SchemaName)
+                    .RegisterStaticTenantsByDataSource(tenants => tenants.Register("red", tenantDataSource));
+            }).StartAsync(TestContext.Current.CancellationToken);
+
+        await host.StopAsync(TestContext.Current.CancellationToken);
+        host.Dispose();
+
+        await shouldStillBeUsable(tenantDataSource);
+    }
+
+    [Fact]
+    public async Task stopping_the_host_disposes_a_data_source_wolverine_built()
+    {
+        var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, SchemaName);
+            }).StartAsync(TestContext.Current.CancellationToken);
+
+        var dataSource = host.Services.GetRequiredService<IMessageStore>().ShouldBeAssignableTo<IMessageDatabase>()!
+            .DataSource;
+
+        await host.StopAsync(TestContext.Current.CancellationToken);
+        host.Dispose();
+
+        await Should.ThrowAsync<ObjectDisposedException>(async () =>
+        {
+            await using var conn = await dataSource.OpenConnectionAsync();
+        });
+    }
+
+    private static async Task shouldStillBeUsable(NpgsqlDataSource dataSource)
+    {
+        await using var command = dataSource.CreateCommand("select 1");
+        (await command.ExecuteScalarAsync()).ShouldBe(1);
+    }
+
+    private static async Task<string> createDatabaseIfNotExists(string databaseName)
+    {
+        await using (var conn = new NpgsqlConnection(Servers.PostgresConnectionString))
+        {
+            await conn.OpenAsync();
+
+            if (!await conn.DatabaseExists(databaseName))
+            {
+                await new DatabaseSpecification().BuildDatabase(conn, databaseName);
+            }
+        }
+
+        return new NpgsqlConnectionStringBuilder(Servers.PostgresConnectionString) { Database = databaseName }
+            .ConnectionString;
+    }
+}