Use inline senders on the tenanted Azure Service Bus path (GH-3826)

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<AzureServiceBusTransport>();
                 transport.Tenants["test"].Transport.ManagementConnectionString =
-                    Servers.AzureServiceBusConnectionString;
+                    Servers.AzureServiceBusManagementConnectionString;
             }).StartAsync(cancellationToken: TestContext.Current.CancellationToken);
 
         var message = new Bug1933Message("Hello from default namespace");

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)
+    {
+    }
+}
+
+/// <summary>
+/// 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.
+/// </summary>
+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<AzureServiceBusTransport>();
+                transport.Tenants["tenant1"].Transport.ManagementConnectionString =
+                    Servers.AzureServiceBusManagementConnectionString;
+
+                opts.PublishMessage<Tenanted3826Message>()
+                    .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<BufferedSendingAgent>();
+
+        var tenanted = agent.Sender.ShouldBeOfType<TenantedSender>();
+
+        // 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<InlineAzureServiceBusSender>();
+        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<Tenanted3826Message>()
+            .Name.ShouldBe("no tenant");
+    }
+}

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,

src/Wolverine/Transports/Sending/TenantedSender.cs

@@ -82,6 +82,20 @@ public void RegisterSender(string tenantId, ISender sender)
         _senders = _senders.AddOrUpdate(tenantId, sender);
     }
 
+    /// <summary>
+    /// The sender used for the untenanted pathway, and for unknown tenant ids when the
+    /// TenantedIdBehavior is FallbackToDefault
+    /// </summary>
+    public ISender DefaultSender => _defaultSender;
+
+    /// <summary>
+    /// All the registered tenant id / sender pairs. Diagnostic only.
+    /// </summary>
+    public IEnumerable<KeyValuePair<string, ISender>> TenantSenders()
+    {
+        return _senders.Enumerate().Select(x => new KeyValuePair<string, ISender>(x.Key, x.Value));
+    }
+
     public bool SupportsNativeScheduledSend => _defaultSender.SupportsNativeScheduledSend;
 
     bool IConditionalNativeScheduling.CanScheduleNatively(Envelope envelope, DateTimeOffset utcNow)