diff --git a/src/Persistence/MySql/MySqlTests/MultiTenancy/queue_tables_are_per_tenant_database.cs b/src/Persistence/MySql/MySqlTests/MultiTenancy/queue_tables_are_per_tenant_database.cs new file mode 100644 index 000000000..362104141 --- /dev/null +++ b/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; + +/// +/// GH-3859. A MySQL schema IS a database, so qualifying the queue tables with the single transport-wide +/// TransportSchemaName resolved every tenant's data source to the *same* physical table: no tenant +/// isolation, and a CountAsync() 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. +/// +[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().Queues[QueueName]; + + /// + /// The structural half: every tenant database must physically own its queue tables. Before the fix + /// they existed only in the single "wolverine_queues" database. + /// + [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"); + } + } + + /// + /// The behavioural half: a tenant's message lands in that tenant's database and nowhere else. + /// + [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); + } + } + + /// + /// 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. + /// + [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 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 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(); + } + } +} diff --git a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueue.cs b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueue.cs index 94d087ec2..0d667e040 100644 --- a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueue.cs +++ b/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 _checkedDatabases = ImHashMap.Empty; + private ImHashMap _queueTables = ImHashMap.Empty; + private ImHashMap _scheduledTables = ImHashMap.Empty; + private readonly string _queueTableName; + private readonly string _scheduledTableName; private readonly Lazy _queueTable; private readonly Lazy _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(() => new QueueTable(Parent, queueTableName)); + _queueTable = new Lazy(() => new QueueTable(Parent, _queueTableName)); _scheduledMessageTable = - new Lazy(() => new ScheduledMessageTable(Parent, scheduledTableName)); + new Lazy(() => 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; + /// + /// GH-3859. MySQL has no schema-inside-database nesting — a schema IS a database — so the single + /// 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. + /// + 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,13 +179,16 @@ public ValueTask SendAsync(Envelope envelope) return _sender!.SendAsync(envelope); } + /// + /// GH-3815. These two sources overlap: MultiTenantedMessageStore.ActiveDatabases() yields + /// Main first, and is only ever assigned alongside + /// Store = mt.Main. Visiting both therefore hit the main database twice — doubling + /// /, which + /// reports as user visible queue depth, and running every schema check against it twice. The + /// SqlServer and Sqlite queues already branch this way. + /// private async ValueTask forEveryDatabase(Func 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()) @@ -151,6 +196,10 @@ private async ValueTask forEveryDatabase(Func act 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 diff --git a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueListener.cs b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueListener.cs index 63d96c6c0..3ce32a2a2 100644 --- a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueListener.cs +++ b/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); diff --git a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueSender.cs b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueSender.cs index 24dd8180b..20b8053ad 100644 --- a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueSender.cs +++ b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueSender.cs @@ -31,9 +31,15 @@ public MySqlQueueSender(MySqlQueue queue, MySqlDataSource dataSource, string? da Destination = MySqlQueue.ToUri(queue.Name, databaseName); - _schemaName = queue.Parent.TransportSchemaName; + // GH-3859: on a multi-tenanted host the queue tables live in each tenant's OWN database, so + // every one of the statements below has to be built against THIS sender's data source rather + // than the transport-wide schema name. + var queueTable = queue.QueueTableFor(dataSource).Identifier; + var scheduledTable = queue.ScheduledTableFor(dataSource).Identifier; + + _schemaName = queue.SchemaFor(dataSource); _moveFromOutgoingToQueueSql = $@" -INSERT INTO {queue.QueueTable.Identifier} ({DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.KeepUntil}) +INSERT INTO {queueTable} ({DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.KeepUntil}) SELECT {DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.DeliverBy} FROM {queue.Parent.MessageStorageSchemaName}.{DatabaseConstants.OutgoingTable} WHERE {DatabaseConstants.Id} = @id; @@ -41,7 +47,7 @@ public MySqlQueueSender(MySqlQueue queue, MySqlDataSource dataSource, string? da "; _moveFromOutgoingToScheduledSql = $@" -INSERT INTO {queue.ScheduledTable.Identifier} ({DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.ExecutionTime}, {DatabaseConstants.KeepUntil}) +INSERT INTO {scheduledTable} ({DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.ExecutionTime}, {DatabaseConstants.KeepUntil}) SELECT {DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, @time, {DatabaseConstants.DeliverBy} FROM {queue.Parent.MessageStorageSchemaName}.{DatabaseConstants.OutgoingTable} WHERE {DatabaseConstants.Id} = @id; @@ -49,10 +55,10 @@ public MySqlQueueSender(MySqlQueue queue, MySqlDataSource dataSource, string? da "; _writeDirectlyToQueueTableSql = - $@"INSERT INTO {queue.QueueTable.Identifier} ({DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.KeepUntil}) VALUES (@id, @body, @type, @expires)"; + $@"INSERT INTO {queueTable} ({DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.KeepUntil}) VALUES (@id, @body, @type, @expires)"; _writeDirectlyToTheScheduledTable = $@" -INSERT INTO {queue.ScheduledTable.Identifier} ({DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.KeepUntil}, {DatabaseConstants.ExecutionTime}) +INSERT INTO {scheduledTable} ({DatabaseConstants.Id}, {DatabaseConstants.Body}, {DatabaseConstants.MessageType}, {DatabaseConstants.KeepUntil}, {DatabaseConstants.ExecutionTime}) VALUES (@id, @body, @type, @expires, @time) ON DUPLICATE KEY UPDATE {DatabaseConstants.Body} = @body, {DatabaseConstants.MessageType} = @type, {DatabaseConstants.KeepUntil} = @expires, {DatabaseConstants.ExecutionTime} = @time ".Trim(); diff --git a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlTransport.cs b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlTransport.cs index 6b30b18ee..0895a2a89 100644 --- a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlTransport.cs +++ b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlTransport.cs @@ -75,8 +75,15 @@ public override async ValueTask ConnectAsync(IWolverineRuntime runtime) foreach (var queue in Queues) { - Store.AddTable(queue.QueueTable); - Store.AddTable(queue.ScheduledTable); + // GH-3859: on a multi-tenanted host the queue tables live in each database rather than in + // one transport-wide schema, and SetupAsync()/EnsureSchemaExists() provision them per data + // source. Registering a single set with the main store here would only ever provision one + // that nothing uses. + if (Databases == null) + { + Store.AddTable(queue.QueueTableFor(Store.MySqlDataSource)); + Store.AddTable(queue.ScheduledTableFor(Store.MySqlDataSource)); + } } MessageStorageSchemaName = Store.SchemaName; diff --git a/src/Persistence/MySql/Wolverine.MySql/Transport/QueueTable.cs b/src/Persistence/MySql/Wolverine.MySql/Transport/QueueTable.cs index 34704cb58..3e87016f5 100644 --- a/src/Persistence/MySql/Wolverine.MySql/Transport/QueueTable.cs +++ b/src/Persistence/MySql/Wolverine.MySql/Transport/QueueTable.cs @@ -6,8 +6,13 @@ namespace Wolverine.MySql.Transport; internal class QueueTable : Table { - public QueueTable(MySqlTransport parent, string tableName) : base( - new DbObjectName(parent.TransportSchemaName, tableName)) + public QueueTable(MySqlTransport parent, string tableName) : this(parent.TransportSchemaName, tableName) + { + } + + // GH-3859: the schema is resolved per data source on a multi-tenanted host, because a MySQL schema + // IS a database and one fixed name would leave every tenant sharing a single physical queue table. + public QueueTable(string schemaName, string tableName) : base(new DbObjectName(schemaName, tableName)) { AddColumn(DatabaseConstants.Id).AsPrimaryKey(); AddColumn(DatabaseConstants.Body, "LONGBLOB").NotNull(); diff --git a/src/Persistence/MySql/Wolverine.MySql/Transport/ScheduledMessageTable.cs b/src/Persistence/MySql/Wolverine.MySql/Transport/ScheduledMessageTable.cs index 2ab070c8f..d7fc58f49 100644 --- a/src/Persistence/MySql/Wolverine.MySql/Transport/ScheduledMessageTable.cs +++ b/src/Persistence/MySql/Wolverine.MySql/Transport/ScheduledMessageTable.cs @@ -6,8 +6,13 @@ namespace Wolverine.MySql.Transport; internal class ScheduledMessageTable : Table { - public ScheduledMessageTable(MySqlTransport settings, string tableName) : base( - new DbObjectName(settings.TransportSchemaName, tableName)) + public ScheduledMessageTable(MySqlTransport settings, string tableName) + : this(settings.TransportSchemaName, tableName) + { + } + + // GH-3859: see the matching comment on QueueTable. + public ScheduledMessageTable(string schemaName, string tableName) : base(new DbObjectName(schemaName, tableName)) { AddColumn(DatabaseConstants.Id).AsPrimaryKey(); AddColumn(DatabaseConstants.Body, "LONGBLOB").NotNull(); diff --git a/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueue.cs b/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueue.cs index 86f8671f5..6f8d6e337 100644 --- a/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueue.cs +++ b/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueue.cs @@ -137,13 +137,16 @@ public ValueTask SendAsync(Envelope envelope) return _sender!.SendAsync(envelope); } + /// + /// GH-3815. These two sources overlap: MultiTenantedMessageStore.ActiveDatabases() yields + /// Main first, and is only ever assigned alongside + /// Store = mt.Main. Visiting both therefore hit the main database twice — doubling + /// /, which + /// reports as user visible queue depth, and running every schema check against it twice. The + /// SqlServer and Sqlite queues already branch this way. + /// private async ValueTask forEveryDatabase(Func action) { - if (Parent?.Store?.OracleDataSource != null) - { - await action(Parent.Store.OracleDataSource, Parent.Store.Name); - } - if (Parent?.Databases != null) { foreach (var database in Parent.Databases.ActiveDatabases().OfType()) @@ -151,6 +154,10 @@ private async ValueTask forEveryDatabase(Func ac await action(database.OracleDataSource, database.Name); } } + else if (Parent?.Store?.OracleDataSource != null) + { + await action(Parent.Store.OracleDataSource, Parent.Store.Name); + } } public ValueTask PurgeAsync(ILogger logger) diff --git a/src/Persistence/PostgresqlTests/MultiTenancy/queue_counts_across_tenant_databases.cs b/src/Persistence/PostgresqlTests/MultiTenancy/queue_counts_across_tenant_databases.cs new file mode 100644 index 000000000..aff35be29 --- /dev/null +++ b/src/Persistence/PostgresqlTests/MultiTenancy/queue_counts_across_tenant_databases.cs @@ -0,0 +1,173 @@ +using IntegrationTests; +using JasperFx.Core; +using JasperFx.Resources; +using Microsoft.Extensions.DependencyInjection; +using Npgsql; +using Shouldly; +using Wolverine; +using Wolverine.ComplianceTests; +using Wolverine.Persistence.Durability; +using Wolverine.Postgresql; +using Wolverine.Postgresql.Transport; +using Wolverine.Runtime; +using Wolverine.Tracking; +using Xunit; + +namespace PostgresqlTests.MultiTenancy; + +/// +/// GH-3815. forEveryDatabase walked Parent.Store and then every entry of +/// Parent.Databases.ActiveDatabases() — but MultiTenantedMessageStore.ActiveDatabases() +/// yields Main first, and Databases is only ever assigned alongside Store = mt.Main. +/// The main database was therefore visited twice, so CountAsync() and ScheduledCountAsync() +/// double counted every row living in it. Those two feed GetAttributesAsync(), so this was +/// user visible queue depth, not just a test concern. +/// +/// The existing multi-tenant coverage misses it because it only ever asserts a count of 0, and +/// zero doubled is still zero. +/// +public class queue_counts_across_tenant_databases : MultiTenancyContext +{ + private const string SchemaName = "queue_counts_tenanted"; + private const string QueueName = "countone"; + + protected override void configureWolverine(WolverineOptions opts) + { + opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, SchemaName) + .EnableMessageTransport(transport => transport.TransportSchemaName(SchemaName)) + .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().ToPostgresqlQueue(QueueName); + + opts.Services.AddResourceSetupOnStartup(); + } + + protected override async Task onStartup() + { + foreach (var connectionString in allConnectionStrings()) + { + await using var conn = new NpgsqlConnection(connectionString); + await conn.OpenAsync(); + try + { + foreach (var table in new[] { $"wolverine_queue_{QueueName}", $"wolverine_queue_{QueueName}_scheduled" }) + { + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"delete from {SchemaName}.{table}"; + try + { + await cmd.ExecuteNonQueryAsync(); + } + catch (PostgresException e) when (e.SqlState == PostgresErrorCodes.UndefinedTable || + e.SqlState == PostgresErrorCodes.InvalidSchemaName) + { + // Nothing provisioned in this database yet, nothing to clean + } + } + } + finally + { + await conn.CloseAsync(); + } + } + } + + private string[] allConnectionStrings() => + [ + Servers.PostgresConnectionString, tenant1ConnectionString, tenant2ConnectionString, tenant3ConnectionString + ]; + + private PostgresqlQueue theQueue => + theHost.GetRuntime().Options.Transports.GetOrCreate().Queues[QueueName]; + + /// + /// A row in the *main* database is the one that gets double counted — an untenanted send lands there. + /// + [Fact] + public async Task does_not_double_count_rows_in_the_main_database() + { + var runtime = theHost.GetRuntime(); + ((MultiTenantedMessageStore)runtime.Storage).ActiveDatabases().Count.ShouldBe(4); + + var immediate = ObjectMother.Envelope(); + immediate.DeliverBy = DateTimeOffset.UtcNow.AddHours(1); + await theQueue.SendAsync(immediate); + + var scheduled = ObjectMother.Envelope(); + scheduled.ScheduleDelay = 1.Hours(); + scheduled.DeliverBy = DateTimeOffset.UtcNow.AddHours(1); + await theQueue.SendAsync(scheduled); + + // Precondition: exactly one row physically exists, in the main database only + (await rowCountAsync(Servers.PostgresConnectionString, $"wolverine_queue_{QueueName}")).ShouldBe(1); + (await rowCountAsync(Servers.PostgresConnectionString, $"wolverine_queue_{QueueName}_scheduled")).ShouldBe(1); + + (await theQueue.CountAsync()).ShouldBe(1); + (await theQueue.ScheduledCountAsync()).ShouldBe(1); + } + + /// + /// And the sum still reaches every tenant database -- the fix must not trade the double count for a + /// missed database. + /// + [Fact] + public async Task still_sums_across_main_and_every_tenant_database() + { + 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, $"wolverine_queue_{QueueName}")).ShouldBe(1); + } + + // One row in each of the four databases, counted once apiece + (await theQueue.CountAsync()).ShouldBe(4); + } + + /// + /// GetAttributesAsync() is the user visible surface -- it reports whatever CountAsync() returns. + /// + [Fact] + public async Task reported_attributes_match_the_physical_row_count() + { + var envelope = ObjectMother.Envelope(); + envelope.DeliverBy = DateTimeOffset.UtcNow.AddHours(1); + await theQueue.SendAsync(envelope); + + var attributes = await theQueue.GetAttributesAsync(); + + attributes["Count"].ShouldBe("1"); + } + + private static async Task rowCountAsync(string connectionString, string tableName) + { + await using var conn = new NpgsqlConnection(connectionString); + await conn.OpenAsync(); + try + { + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"select count(*) from {SchemaName}.{tableName}"; + return (long)(await cmd.ExecuteScalarAsync())!; + } + finally + { + await conn.CloseAsync(); + } + } +} diff --git a/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueue.cs b/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueue.cs index a6e906416..37c2f4f45 100644 --- a/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueue.cs +++ b/src/Persistence/Wolverine.Postgresql/Transport/PostgresqlQueue.cs @@ -144,13 +144,16 @@ public ValueTask SendAsync(Envelope envelope) return _sender!.SendAsync(envelope); } + /// + /// GH-3815. These two sources overlap: MultiTenantedMessageStore.ActiveDatabases() yields + /// Main first, and is only ever assigned alongside + /// Store = mt.Main. Visiting both therefore hit the main database twice — doubling + /// /, which + /// reports as user visible queue depth, and running every schema check against it twice. The + /// SqlServer and Sqlite queues already branch this way. + /// private async ValueTask forEveryDatabase(Func action) { - if (Parent?.Store?.NpgsqlDataSource != null) - { - await action(Parent.Store.NpgsqlDataSource, Parent.Store.Identifier); - } - if (Parent?.Databases != null) { foreach (var database in Parent.Databases.ActiveDatabases().OfType()) @@ -158,6 +161,10 @@ private async ValueTask forEveryDatabase(Func ac await action(database.NpgsqlDataSource, database.Identifier); } } + else if (Parent?.Store?.NpgsqlDataSource != null) + { + await action(Parent.Store.NpgsqlDataSource, Parent.Store.Identifier); + } } public ValueTask PurgeAsync(ILogger logger)