diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/Bugs/Bug_1933_multi_tenant_conventional_routing.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/Bugs/Bug_1933_multi_tenant_conventional_routing.cs index c7a6fc0e3..7dfad90a1 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/Bugs/Bug_1933_multi_tenant_conventional_routing.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/Bugs/Bug_1933_multi_tenant_conventional_routing.cs @@ -16,12 +16,6 @@ public static void Handle(Bug1933Message message) } } -// NOT flaky. Re-measured 2026-08-02 after the entity-name sanitizing fix: now 1 of 2, down from 2 of -// 2, and 5.5m for the class. The broker starts fine now; the survivor is -// should_receive_message_when_published_without_tenant_id, which fails as a TrackedSession timeout -// after 4m28s -- the message is Sent and never Received. That is real multi-tenant routing -// behaviour, not provisioning. Tracked as GH-3826. -[Trait("Category", "Flaky")] public class Bug_1933_multi_tenant_conventional_routing : IAsyncLifetime { public async ValueTask InitializeAsync() =>await ValueTask.CompletedTask; @@ -55,10 +49,13 @@ public async Task should_receive_message_when_published_without_tenant_id() .AddTenantByConnectionString("test", Servers.AzureServiceBusConnectionString) .UseConventionalRouting(); - // Set the tenant's management connection string for the emulator + // Set the tenant's management connection string for the emulator. This has to be the + // *management* endpoint -- pointing it at the AMQP endpoint made every tenant + // provisioning call hang until its own timeout, which is where this test's old + // 4m+ runtime came from. var transport = opts.Transports.GetOrCreate(); transport.Tenants["test"].Transport.ManagementConnectionString = - Servers.AzureServiceBusConnectionString; + Servers.AzureServiceBusManagementConnectionString; }).StartAsync(cancellationToken: TestContext.Current.CancellationToken); var message = new Bug1933Message("Hello from default namespace"); diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/tenanted_senders_are_inline_3826.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/tenanted_senders_are_inline_3826.cs new file mode 100644 index 000000000..b0406995d --- /dev/null +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/tenanted_senders_are_inline_3826.cs @@ -0,0 +1,90 @@ +using IntegrationTests; +using JasperFx.Core; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.AzureServiceBus.Internal; +using Wolverine.Runtime; +using Wolverine.Tracking; +using Wolverine.Transports.Sending; +using Xunit; + +namespace Wolverine.AzureServiceBus.Tests; + +public record Tenanted3826Message(string Name); + +public static class Tenanted3826MessageHandler +{ + public static void Handle(Tenanted3826Message message) + { + } +} + +/// +/// GH-3826. TenantedSender deliberately does not implement ISenderRequiresCallback (GH-2361), so +/// EndpointCollection never calls RegisterCallback on the senders underneath it. A BatchedSender in +/// that position therefore throws "This sender has not been registered." on every batch, and a +/// tenanted Azure Service Bus endpoint could never send anything at all -- not even on the +/// untenanted default pathway. Every sender under a TenantedSender has to be fire-and-forget. +/// +public class tenanted_senders_are_inline_3826 : IAsyncLifetime +{ + private IHost _host = null!; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.ServiceName = "tenanted3826"; + opts.Policies.DisableConventionalLocalRouting(); + + opts.UseAzureServiceBusTesting() + .AutoPurgeOnStartup() + .AddTenantByConnectionString("tenant1", Servers.AzureServiceBusConnectionString); + + var transport = opts.Transports.GetOrCreate(); + transport.Tenants["tenant1"].Transport.ManagementConnectionString = + Servers.AzureServiceBusManagementConnectionString; + + opts.PublishMessage() + .ToAzureServiceBusQueue("tenanted-3826") + .BufferedInMemory(); + + opts.ListenToAzureServiceBusQueue("tenanted-3826").BufferedInMemory(); + }).StartAsync(cancellationToken: TestContext.Current.CancellationToken); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + [Fact] + public void every_sender_under_the_tenanted_sender_is_fire_and_forget() + { + var runtime = _host.GetRuntime(); + var agent = runtime.Endpoints.GetOrBuildSendingAgent(new Uri("asb://queue/tenanted-3826")) + .ShouldBeOfType(); + + var tenanted = agent.Sender.ShouldBeOfType(); + + // The default fallback -- the pathway an untenanted publish takes -- and the tenant sender + // both have to be inline. A BatchedSender here has no callback and fails every send. + tenanted.DefaultSender.ShouldBeOfType(); + tenanted.TenantSenders().Select(x => x.Value) + .ShouldAllBe(x => x is InlineAzureServiceBusSender); + } + + [Fact] + public async Task can_actually_send_without_a_tenant_id() + { + var session = await _host.TrackActivity() + .IncludeExternalTransports() + .Timeout(30.Seconds()) + .SendMessageAndWaitAsync(new Tenanted3826Message("no tenant")); + + session.Received.SingleMessage() + .Name.ShouldBe("no tenant"); + } +} diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Sending.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Sending.cs index b3694a6f4..a04e92b85 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Sending.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Sending.cs @@ -10,23 +10,20 @@ public partial class AzureServiceBusTransport { internal ISender CreateSender(IWolverineRuntime runtime, AzureServiceBusTopic topic) { - var mapper = topic.BuildMapper(runtime); - - var defaultSender = buildSenderForTopic(runtime, topic, mapper); - + // GH-3826: when tenants are in play, every sender underneath the TenantedSender -- the + // default fallback included -- has to be the fire-and-forget inline sender. TenantedSender + // deliberately does not implement ISenderRequiresCallback (GH-2361), so EndpointCollection + // never calls RegisterCallback on the senders beneath it, and a BatchedSender there throws + // "This sender has not been registered." on every single batch. Same reasoning and same + // shape as Redis, MQTT, and Pub/Sub. if (Tenants.Any() && topic.TenancyBehavior == TenancyBehavior.TenantAware) { - var tenantedSender = new TenantedSender(topic.Uri, TenantedIdBehavior, defaultSender); - foreach (var tenant in Tenants) - { - var sender = tenant.Transport.buildSenderForTopic(runtime, topic, mapper); - tenantedSender.RegisterSender(tenant.TenantId, sender); - } - - return tenantedSender; + return BuildInlineSenderForTopic(runtime, topic); } - return defaultSender; + var mapper = topic.BuildMapper(runtime); + + return buildSenderForTopic(runtime, topic, mapper); } private ISender buildSenderForTopic(IWolverineRuntime runtime, AzureServiceBusTopic topic, @@ -106,22 +103,16 @@ private ISender buildInlineSenderForQueue(IWolverineRuntime runtime, AzureServic internal ISender BuildSenderForQueue(IWolverineRuntime runtime, AzureServiceBusQueue queue) { - var mapper = queue.BuildMapper(runtime); - var defaultSender = buildSenderForQueue(runtime, queue, mapper); - + // GH-3826: see CreateSender(topic) -- a BatchedSender underneath a TenantedSender never + // receives its ISenderCallback and fails every batch, so the tenanted path is inline only. if (Tenants.Any() && queue.TenancyBehavior == TenancyBehavior.TenantAware) { - var tenantedSender = new TenantedSender(queue.Uri, TenantedIdBehavior, defaultSender); - foreach (var tenant in Tenants) - { - var sender = tenant.Transport.buildSenderForQueue(runtime, queue, mapper); - tenantedSender.RegisterSender(tenant.TenantId, sender); - } - - return tenantedSender; + return BuildInlineSenderForQueue(runtime, queue); } - - return defaultSender; + + var mapper = queue.BuildMapper(runtime); + + return buildSenderForQueue(runtime, queue, mapper); } private ISender buildSenderForQueue(IWolverineRuntime runtime, AzureServiceBusQueue queue, diff --git a/src/Wolverine/Transports/Sending/TenantedSender.cs b/src/Wolverine/Transports/Sending/TenantedSender.cs index 6c5236577..6e51c4909 100644 --- a/src/Wolverine/Transports/Sending/TenantedSender.cs +++ b/src/Wolverine/Transports/Sending/TenantedSender.cs @@ -82,6 +82,20 @@ public void RegisterSender(string tenantId, ISender sender) _senders = _senders.AddOrUpdate(tenantId, sender); } + /// + /// The sender used for the untenanted pathway, and for unknown tenant ids when the + /// TenantedIdBehavior is FallbackToDefault + /// + public ISender DefaultSender => _defaultSender; + + /// + /// All the registered tenant id / sender pairs. Diagnostic only. + /// + public IEnumerable> TenantSenders() + { + return _senders.Enumerate().Select(x => new KeyValuePair(x.Key, x.Value)); + } + public bool SupportsNativeScheduledSend => _defaultSender.SupportsNativeScheduledSend; bool IConditionalNativeScheduling.CanScheduleNatively(Envelope envelope, DateTimeOffset utcNow)