MySQL: resolve queue tables per tenant database (GH-3859)

src/Persistence/MySql/MySqlTests/MultiTenancy/queue_tables_are_per_tenant_database.cs

@@ -0,0 +1,189 @@
+using IntegrationTests;
+using JasperFx.Core;
+using JasperFx.Resources;
+using Microsoft.Extensions.DependencyInjection;
+using MySqlConnector;
+using Shouldly;
+using Wolverine;
+using Wolverine.ComplianceTests;
+using Wolverine.MySql;
+using Wolverine.MySql.Transport;
+using Wolverine.Persistence.Durability;
+using Wolverine.Runtime;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace MySqlTests.MultiTenancy;
+
+/// <summary>
+/// GH-3859. A MySQL schema IS a database, so qualifying the queue tables with the single transport-wide
+/// <c>TransportSchemaName</c> resolved every tenant's data source to the *same* physical table: no tenant
+/// isolation, and a <c>CountAsync()</c> that multiplied the true row count by the number of tenants.
+/// On a multi-tenanted host each data source now resolves its queue tables inside its own database.
+/// </summary>
+[Collection("mysql")]
+public class queue_tables_are_per_tenant_database : MySqlMultiTenancyContext
+{
+    private const string SchemaName = "queue_per_tenant";
+    private const string QueueName = "pertenant";
+
+    protected override void configureWolverine(WolverineOptions opts)
+    {
+        opts.PersistMessagesWithMySql(Servers.MySqlConnectionString, SchemaName)
+            .EnableMessageTransport()
+            .RegisterStaticTenants(tenants =>
+            {
+                tenants.Register("red", tenant1ConnectionString);
+                tenants.Register("blue", tenant2ConnectionString);
+                tenants.Register("green", tenant3ConnectionString);
+            });
+
+        // Subscriber only -- no listener, so nothing drains the queue out from under the assertions.
+        opts.PublishAllMessages().ToMySqlQueue(QueueName);
+
+        opts.Services.AddResourceSetupOnStartup();
+    }
+
+    protected override async Task onStartup()
+    {
+        foreach (var connectionString in allConnectionStrings())
+        {
+            await using var conn = new MySqlConnection(connectionString);
+            await conn.OpenAsync();
+            try
+            {
+                foreach (var table in new[] { QueueTableName, ScheduledTableName })
+                {
+                    await using var cmd = conn.CreateCommand();
+                    cmd.CommandText = $"delete from {table}";
+                    try
+                    {
+                        await cmd.ExecuteNonQueryAsync();
+                    }
+                    catch (MySqlException)
+                    {
+                        // Nothing provisioned in this database yet, nothing to clean
+                    }
+                }
+            }
+            finally
+            {
+                await conn.CloseAsync();
+            }
+        }
+    }
+
+    private const string QueueTableName = $"wolverine_queue_{QueueName}";
+    private const string ScheduledTableName = $"wolverine_queue_{QueueName}_scheduled";
+
+    private string[] allConnectionStrings() =>
+    [
+        Servers.MySqlConnectionString, tenant1ConnectionString, tenant2ConnectionString, tenant3ConnectionString
+    ];
+
+    private MySqlQueue theQueue =>
+        theHost.GetRuntime().Options.Transports.GetOrCreate<MySqlTransport>().Queues[QueueName];
+
+    /// <summary>
+    /// The structural half: every tenant database must physically own its queue tables. Before the fix
+    /// they existed only in the single "wolverine_queues" database.
+    /// </summary>
+    [Fact]
+    public async Task every_tenant_database_owns_its_own_queue_tables()
+    {
+        ((MultiTenantedMessageStore)theHost.GetRuntime().Storage).ActiveDatabases().Count.ShouldBe(4);
+
+        foreach (var connectionString in allConnectionStrings())
+        {
+            var database = new MySqlConnectionStringBuilder(connectionString).Database;
+
+            (await tableExistsAsync(connectionString, database, QueueTableName))
+                .ShouldBeTrue($"Expected {database}.{QueueTableName} to exist");
+            (await tableExistsAsync(connectionString, database, ScheduledTableName))
+                .ShouldBeTrue($"Expected {database}.{ScheduledTableName} to exist");
+        }
+    }
+
+    /// <summary>
+    /// The behavioural half: a tenant's message lands in that tenant's database and nowhere else.
+    /// </summary>
+    [Fact]
+    public async Task a_tenants_message_lands_only_in_that_tenants_database()
+    {
+        var envelope = ObjectMother.Envelope();
+        envelope.TenantId = "red";
+        envelope.DeliverBy = DateTimeOffset.UtcNow.AddHours(1);
+        await theQueue.SendAsync(envelope);
+
+        (await rowCountAsync(tenant1ConnectionString)).ShouldBe(1);
+
+        foreach (var other in new[] { Servers.MySqlConnectionString, tenant2ConnectionString, tenant3ConnectionString })
+        {
+            (await rowCountAsync(other)).ShouldBe(0);
+        }
+    }
+
+    /// <summary>
+    /// And with one row per database the sum is the true total, not a multiple of it. This is what
+    /// GetAttributesAsync() reports as queue depth.
+    /// </summary>
+    [Fact]
+    public async Task counts_sum_across_databases_instead_of_multiplying()
+    {
+        var untenanted = ObjectMother.Envelope();
+        untenanted.DeliverBy = DateTimeOffset.UtcNow.AddHours(1);
+        await theQueue.SendAsync(untenanted);
+
+        foreach (var tenantId in new[] { "red", "blue", "green" })
+        {
+            var envelope = ObjectMother.Envelope();
+            envelope.TenantId = tenantId;
+            envelope.DeliverBy = DateTimeOffset.UtcNow.AddHours(1);
+            await theQueue.SendAsync(envelope);
+        }
+
+        foreach (var connectionString in allConnectionStrings())
+        {
+            (await rowCountAsync(connectionString)).ShouldBe(1);
+        }
+
+        (await theQueue.CountAsync()).ShouldBe(4);
+        (await theQueue.GetAttributesAsync())["Count"].ShouldBe("4");
+    }
+
+    private static async Task<bool> tableExistsAsync(string connectionString, string database, string tableName)
+    {
+        await using var conn = new MySqlConnection(connectionString);
+        await conn.OpenAsync();
+        try
+        {
+            await using var cmd = conn.CreateCommand();
+            cmd.CommandText =
+                "select count(*) from information_schema.tables where table_schema = @schema and table_name = @table";
+            cmd.Parameters.AddWithValue("schema", database);
+            cmd.Parameters.AddWithValue("table", tableName);
+
+            return Convert.ToInt64(await cmd.ExecuteScalarAsync()) > 0;
+        }
+        finally
+        {
+            await conn.CloseAsync();
+        }
+    }
+
+    private static async Task<long> rowCountAsync(string connectionString)
+    {
+        await using var conn = new MySqlConnection(connectionString);
+        await conn.OpenAsync();
+        try
+        {
+            await using var cmd = conn.CreateCommand();
+            cmd.CommandText = $"select count(*) from {QueueTableName}";
+            return Convert.ToInt64(await cmd.ExecuteScalarAsync());
+        }
+        finally
+        {
+            await conn.CloseAsync();
+        }
+    }
+}

src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueue.cs

@@ -23,6 +23,10 @@ internal static Uri ToUri(string name, string? databaseName)
     private bool _hasInitialized;
     private IMySqlQueueSender? _sender;
     private ImHashMap<string, bool> _checkedDatabases = ImHashMap<string, bool>.Empty;
+    private ImHashMap<string, QueueTable> _queueTables = ImHashMap<string, QueueTable>.Empty;
+    private ImHashMap<string, ScheduledMessageTable> _scheduledTables = ImHashMap<string, ScheduledMessageTable>.Empty;
+    private readonly string _queueTableName;
+    private readonly string _scheduledTableName;
     private readonly Lazy<QueueTable> _queueTable;
     private readonly Lazy<ScheduledMessageTable> _scheduledMessageTable;
 
@@ -31,17 +35,17 @@ public MySqlQueue(string name, MySqlTransport parent, EndpointRole role = Endpoi
         base(ToUri(name, databaseName), role)
     {
         Parent = parent;
-        var queueTableName = $"wolverine_queue_{name}";
-        var scheduledTableName = $"wolverine_queue_{name}_scheduled";
+        _queueTableName = $"wolverine_queue_{name}";
+        _scheduledTableName = $"wolverine_queue_{name}_scheduled";
 
         Mode = EndpointMode.Durable;
         Name = name;
         EndpointName = name;
         BrokerRole = "queue";
 
-        _queueTable = new Lazy<QueueTable>(() => new QueueTable(Parent, queueTableName));
+        _queueTable = new Lazy<QueueTable>(() => new QueueTable(Parent, _queueTableName));
         _scheduledMessageTable =
-            new Lazy<ScheduledMessageTable>(() => new ScheduledMessageTable(Parent, scheduledTableName));
+            new Lazy<ScheduledMessageTable>(() => new ScheduledMessageTable(Parent, _scheduledTableName));
     }
 
     public string Name { get; }
@@ -52,6 +56,44 @@ public MySqlQueue(string name, MySqlTransport parent, EndpointRole role = Endpoi
 
     internal Table ScheduledTable => _scheduledMessageTable.Value;
 
+    /// <summary>
+    /// GH-3859. MySQL has no schema-inside-database nesting — a schema IS a database — so the single
+    /// <see cref="MySqlTransport.TransportSchemaName"/> that works for the other providers resolves to one
+    /// physical database for every tenant, and all tenants end up sharing one queue table. On a
+    /// multi-tenanted host each data source therefore has to resolve its queue tables inside its *own*
+    /// database. Single database hosts are unaffected and keep using TransportSchemaName.
+    /// </summary>
+    internal string SchemaFor(MySqlDataSource source)
+    {
+        if (Parent.Databases == null) return Parent.TransportSchemaName;
+
+        var database = new MySqlConnectionStringBuilder(source.ConnectionString).Database;
+
+        return database.IsEmpty() ? Parent.TransportSchemaName : database;
+    }
+
+    internal QueueTable QueueTableFor(MySqlDataSource source)
+    {
+        var schema = SchemaFor(source);
+        if (_queueTables.TryFind(schema, out var table)) return table;
+
+        table = new QueueTable(schema, _queueTableName);
+        _queueTables = _queueTables.AddOrUpdate(schema, table);
+
+        return table;
+    }
+
+    internal ScheduledMessageTable ScheduledTableFor(MySqlDataSource source)
+    {
+        var schema = SchemaFor(source);
+        if (_scheduledTables.TryFind(schema, out var table)) return table;
+
+        table = new ScheduledMessageTable(schema, _scheduledTableName);
+        _scheduledTables = _scheduledTables.AddOrUpdate(schema, table);
+
+        return table;
+    }
+
     protected override bool supportsMode(EndpointMode mode)
     {
         return mode == EndpointMode.Durable || mode == EndpointMode.BufferedInMemory;
@@ -137,20 +179,27 @@ public ValueTask SendAsync(Envelope envelope)
         return _sender!.SendAsync(envelope);
     }
 
+    /// <summary>
+    /// GH-3815. These two sources overlap: <c>MultiTenantedMessageStore.ActiveDatabases()</c> yields
+    /// <c>Main</c> first, and <see cref="MySqlTransport.Databases"/> is only ever assigned alongside
+    /// <c>Store = mt.Main</c>. Visiting both therefore hit the main database twice — doubling
+    /// <see cref="CountAsync"/>/<see cref="ScheduledCountAsync"/>, which <see cref="GetAttributesAsync"/>
+    /// reports as user visible queue depth, and running every schema check against it twice. The
+    /// SqlServer and Sqlite queues already branch this way.
+    /// </summary>
     private async ValueTask forEveryDatabase(Func<MySqlDataSource, string, Task> action)
     {
-        if (Parent?.Store?.MySqlDataSource != null)
-        {
-            await action(Parent.Store.MySqlDataSource, Parent.Store.Identifier);
-        }
-
         if (Parent?.Databases != null)
         {
             foreach (var database in Parent.Databases.ActiveDatabases().OfType<MySqlMessageStore>())
             {
                 await action(database.MySqlDataSource, database.Identifier);
             }
         }
+        else if (Parent?.Store?.MySqlDataSource != null)
+        {
+            await action(Parent.Store.MySqlDataSource, Parent.Store.Identifier);
+        }
     }
 
     public ValueTask PurgeAsync(ILogger logger)
@@ -161,11 +210,11 @@ public ValueTask PurgeAsync(ILogger logger)
             try
             {
                 await using var cmd1 = conn.CreateCommand();
-                cmd1.CommandText = $"DELETE FROM {QueueTable.Identifier.QualifiedName}";
+                cmd1.CommandText = $"DELETE FROM {QueueTableFor(source).Identifier.QualifiedName}";
                 await cmd1.ExecuteNonQueryAsync();
 
                 await using var cmd2 = conn.CreateCommand();
-                cmd2.CommandText = $"DELETE FROM {ScheduledTable.Identifier.QualifiedName}";
+                cmd2.CommandText = $"DELETE FROM {ScheduledTableFor(source).Identifier.QualifiedName}";
                 await cmd2.ExecuteNonQueryAsync();
             }
             finally
@@ -192,14 +241,14 @@ await forEveryDatabase(async (source, _) =>
             await using var conn = await source.OpenConnectionAsync();
             try
             {
-                var queueDelta = await QueueTable.FindDeltaAsync(conn);
+                var queueDelta = await QueueTableFor(source).FindDeltaAsync(conn);
                 if (queueDelta.HasChanges())
                 {
                     returnValue = false;
                     return;
                 }
 
-                var scheduledDelta = await ScheduledTable.FindDeltaAsync(conn);
+                var scheduledDelta = await ScheduledTableFor(source).FindDeltaAsync(conn);
 
                 returnValue = returnValue && !scheduledDelta.HasChanges();
             }
@@ -218,8 +267,8 @@ await forEveryDatabase(async (source, _) =>
         {
             await using var conn = await source.OpenConnectionAsync();
 
-            await QueueTable.DropAsync(conn);
-            await ScheduledTable.DropAsync(conn);
+            await QueueTableFor(source).DropAsync(conn);
+            await ScheduledTableFor(source).DropAsync(conn);
 
             await conn.CloseAsync();
         });
@@ -245,8 +294,8 @@ private async Task applySchemaChangesAsync(MySqlDataSource source, string identi
     {
         await using (var conn = await source.OpenConnectionAsync())
         {
-            await QueueTable.ApplyChangesAsync(conn);
-            await ScheduledTable.ApplyChangesAsync(conn);
+            await QueueTableFor(source).ApplyChangesAsync(conn);
+            await ScheduledTableFor(source).ApplyChangesAsync(conn);
 
             await conn.CloseAsync();
         }
@@ -264,7 +313,7 @@ await forEveryDatabase(async (source, _) =>
             try
             {
                 await using var cmd = conn.CreateCommand();
-                cmd.CommandText = $"SELECT COUNT(*) FROM {QueueTable.Identifier.QualifiedName}";
+                cmd.CommandText = $"SELECT COUNT(*) FROM {QueueTableFor(source).Identifier.QualifiedName}";
                 count += Convert.ToInt64(await cmd.ExecuteScalarAsync());
             }
             finally
@@ -285,7 +334,7 @@ await forEveryDatabase(async (source, _) =>
             try
             {
                 await using var cmd = conn.CreateCommand();
-                cmd.CommandText = $"SELECT COUNT(*) FROM {ScheduledTable.Identifier.QualifiedName}";
+                cmd.CommandText = $"SELECT COUNT(*) FROM {ScheduledTableFor(source).Identifier.QualifiedName}";
                 count += Convert.ToInt64(await cmd.ExecuteScalarAsync());
             }
             finally

src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueListener.cs

@@ -43,8 +43,10 @@ public MySqlQueueListener(MySqlQueue queue, IWolverineRuntime runtime, IReceiver
 
         _sender = new MySqlQueueSender(queue, _dataSource, databaseName);
 
-        _queueTableName = _queue.QueueTable.Identifier.QualifiedName;
-        _scheduledTableName = _queue.ScheduledTable.Identifier.QualifiedName;
+        // GH-3859: resolved against this listener's own data source -- on a multi-tenanted host each
+        // tenant's queue tables live in that tenant's database.
+        _queueTableName = _queue.QueueTableFor(dataSource).Identifier.QualifiedName;
+        _scheduledTableName = _queue.ScheduledTableFor(dataSource).Identifier.QualifiedName;
         _schemaName = _queue.Parent.MessageStorageSchemaName;
 
         // MySQL doesn't have CTID, so we use a different approach:
@@ -424,7 +426,7 @@ DELETE FROM {_queueTableName}
                 .ExecuteNonQueryAsync(cancellationToken);
 
             await conn.CreateCommand($@"
-DELETE FROM {_queue.ScheduledTable.Identifier}
+DELETE FROM {_scheduledTableName}
 WHERE {DatabaseConstants.KeepUntil} IS NOT NULL
 AND {DatabaseConstants.KeepUntil} <= UTC_TIMESTAMP(6)")
                 .ExecuteNonQueryAsync(cancellationToken);