diff --git a/docs/guide/messaging/transports/sqs/index.md b/docs/guide/messaging/transports/sqs/index.md index aa6218014..dc1625103 100644 --- a/docs/guide/messaging/transports/sqs/index.md +++ b/docs/guide/messaging/transports/sqs/index.md @@ -24,7 +24,7 @@ var host = await Host.CreateDefaultBuilder() .AutoPurgeOnStartup(); }).StartAsync(); ``` -snippet source | anchor +snippet source | anchor @@ -54,7 +54,7 @@ builder.UseWolverine(opts => using var host = builder.Build(); await host.StartAsync(); ``` -snippet source | anchor +snippet source | anchor @@ -72,7 +72,7 @@ var host = await Host.CreateDefaultBuilder() opts.UseAmazonSqsTransportLocally(); }).StartAsync(); ``` -snippet source | anchor +snippet source | anchor And lastly, if you want to explicitly supply an access and secret key for your credentials to SQS, you can use this syntax: @@ -106,7 +106,7 @@ builder.UseWolverine(opts => using var host = builder.Build(); await host.StartAsync(); ``` -snippet source | anchor +snippet source | anchor ## Connecting to Multiple Brokers @@ -143,12 +143,104 @@ using var host = await Host.CreateDefaultBuilder() // Other configuration }).StartAsync(); ``` -snippet source | anchor +snippet source | anchor Note that the `Uri` scheme within Wolverine for any endpoints from a "named" Amazon SQS broker is the name that you supply for the broker. So in the example above, you might see `Uri` values for `emea://colors` or `americas://red`. +## Multi-Tenancy with a Broker per Tenant + +Named brokers (above) are a *static* topology: you pin specific endpoints to a specific broker by name at +configuration time. **Broker-per-tenant** is different — it is *runtime* routing. You declare one shared queue +topology, and each tenant is served by its **own dedicated SQS connection** (a distinct AWS account/credentials, +region, or `ServiceURL`). Which connection a message goes to (and which connection an inbound message came from) is +decided at runtime by the message's [tenant id](/guide/handlers/multi-tenancy), typically set through +`DeliveryOptions.TenantId`: + + + +```cs +using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + // The "default" / shared SQS connection + opts.UseAmazonSqsTransport(config => + { + config.RegionEndpoint = RegionEndpoint.USEast1; + }) + .AutoProvision() + + // How should Wolverine route a message whose TenantId is null or + // unknown? FallbackToDefault (the default) uses the shared connection; + // TenantIdRequired throws; IgnoreUnknownTenants silently drops it. + .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault) + + // Each tenant gets its OWN dedicated SQS connection, but shares the + // queue topology declared below. This tenant inherits the parent's + // AWS credentials, and just re-points at its own region. + .AddTenant("tenant-west", config => + { + config.RegionEndpoint = RegionEndpoint.USWest2; + }) + + // Or give the tenant its own dedicated AWS account by supplying + // its own credentials (optionally with its own region/endpoint too): + .AddTenant("tenant-eu", new BasicAWSCredentials("tenant-eu-key", "tenant-eu-secret"), + config => + { + config.RegionEndpoint = RegionEndpoint.EUWest1; + }); + + // One shared topology; messages are routed to the right connection at + // runtime by Envelope.TenantId (e.g. new DeliveryOptions { TenantId = "tenant-west" }). + opts.PublishMessage().ToSqsQueue("colors"); + opts.ListenToSqsQueue("colors"); + }).StartAsync(); +``` +snippet source | anchor + + +To route a specific message to a tenant's connection, stamp the tenant id on the send: + +```csharp +await bus.SendAsync(new ColorMessage("blue"), new DeliveryOptions { TenantId = "tenant-west" }); +``` + +Wolverine wraps the outbound endpoint in a `TenantedSender` that dispatches on `Envelope.TenantId`, and builds a +compound listener that runs one poller per tenant connection — each inbound envelope is stamped with the tenant id +of the connection it was consumed from. This mirrors the +[RabbitMQ](/guide/messaging/transports/rabbitmq/multi-tenancy) and [Azure Service Bus](/guide/messaging/transports/azureservicebus/multitenancy) +broker-per-tenant support. + +::: tip Named broker vs. broker-per-tenant +Use a **named broker** when a *fixed set of endpoints* should always talk to a *specific* broker. Use +**broker-per-tenant** when the *same logical endpoints* should be transparently routed to a *different connection per +tenant* based on the runtime tenant id. They are independent features and can be combined. +::: + +### How a tenant connection is seeded + +Each tenant owns its own child transport — its own `IAmazonSQS` client *and* its own queue-url cache (which is why +tenants can safely share a queue name without their cached `QueueUrl` values colliding). At startup Wolverine seeds +the tenant's connection from the parent's — AWS credentials, region/`ServiceURL`, `AuthenticationRegion`, +auto-provisioning, and the dead-letter-queue configuration — and *then* applies your `AddTenant(...)` overrides, so a +tenant only re-points the axes it actually sets and inherits everything else. + +### Choosing the unknown-tenant behavior + +`TenantIdBehavior(...)` controls what happens when a message has a null or unregistered tenant id: + +* `FallbackToDefault` (the default) — route it to the shared/default connection (the one passed to `UseAmazonSqsTransport`). +* `TenantIdRequired` — throw; every message must carry a known tenant id. +* `IgnoreUnknownTenants` — silently drop the message. + +### Auto-provisioning per connection + +When [`AutoProvision()`](#) is enabled, Wolverine provisions the shared queue topology — including each queue's dead +letter queue — on **every** tenant connection, not just the default one, since each is an independent broker. A tenant +listener's dead-lettered messages are likewise sent to the dead letter queue on that tenant's own connection. + ## Identifier Prefixing for Shared Brokers When sharing a single AWS account or SQS namespace between multiple developers or development environments, you can use `PrefixIdentifiers()` to automatically prepend a prefix to every queue name created by Wolverine. This helps isolate cloud resources for each developer or environment: diff --git a/src/Transports/AWS/Wolverine.AmazonSqs.Tests/AmazonSqsPerTenantConfigurationTests.cs b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/AmazonSqsPerTenantConfigurationTests.cs new file mode 100644 index 000000000..f7684401d --- /dev/null +++ b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/AmazonSqsPerTenantConfigurationTests.cs @@ -0,0 +1,128 @@ +using Amazon; +using Amazon.Runtime; +using JasperFx.Core; +using Shouldly; +using Wolverine.AmazonSqs.Internal; +using Wolverine.Configuration; +using Xunit; + +namespace Wolverine.AmazonSqs.Tests; + +// Broker-per-tenant (GH-3304) unit coverage that needs NO SQS broker/LocalStack: it exercises the tenant +// registration / compile / sender-resolution wiring, not live send/receive. Live tenant routing behavior is +// covered by AmazonSqsPerTenantConnectionTests. +public class AmazonSqsPerTenantConfigurationTests +{ + [Fact] + public void add_tenant_registers_a_tenant_with_its_own_region() + { + var options = new WolverineOptions(); + var sqs = options.UseAmazonSqsTransport(c => c.RegionEndpoint = RegionEndpoint.USEast1); + sqs.AddTenant("tenantB", c => c.RegionEndpoint = RegionEndpoint.USWest2); + + var transport = options.Transports.GetOrCreate(); + transport.Tenants.Count().ShouldBe(1); + + var tenant = transport.Tenants["tenantB"]; + tenant.Compile(transport, null!); + + tenant.Transport.Config.RegionEndpoint.ShouldBe(RegionEndpoint.USWest2); + } + + [Fact] + public void add_tenant_with_credentials_sets_a_dedicated_credential_source() + { + var options = new WolverineOptions(); + var sqs = options.UseAmazonSqsTransport(c => c.RegionEndpoint = RegionEndpoint.USEast1); + + var credentials = new BasicAWSCredentials("tenant-key", "tenant-secret"); + sqs.AddTenant("tenantB", credentials, c => c.RegionEndpoint = RegionEndpoint.EUWest1); + + var transport = options.Transports.GetOrCreate(); + var tenant = transport.Tenants["tenantB"]; + tenant.Compile(transport, null!); + + var credentialSource = tenant.Transport.CredentialSource; + credentialSource.ShouldNotBeNull(); + credentialSource(null!).ShouldBeSameAs(credentials); + tenant.Transport.Config.RegionEndpoint.ShouldBe(RegionEndpoint.EUWest1); + } + + [Fact] + public void tenant_inherits_parent_credentials_when_not_overridden() + { + var options = new WolverineOptions(); + var parentCredentials = new BasicAWSCredentials("parent-key", "parent-secret"); + var sqs = options.UseAmazonSqsTransport(c => c.RegionEndpoint = RegionEndpoint.USEast1); + sqs.Credentials(parentCredentials); + sqs.AddTenant("tenantB", c => c.RegionEndpoint = RegionEndpoint.USWest2); + + var transport = options.Transports.GetOrCreate(); + var tenant = transport.Tenants["tenantB"]; + tenant.Compile(transport, null!); + + // Inherited the parent credential source (dedicated-region tenant, shared account)... + tenant.Transport.CredentialSource.ShouldBeSameAs(transport.CredentialSource); + // ...but kept its own region. + tenant.Transport.Config.RegionEndpoint.ShouldBe(RegionEndpoint.USWest2); + } + + [Fact] + public void tenant_inherits_parent_region_and_dlq_behavior_when_it_sets_neither_service_url_nor_region() + { + var options = new WolverineOptions(); + var sqs = options.UseAmazonSqsTransport(c => c.RegionEndpoint = RegionEndpoint.USEast1); + sqs.DefaultDeadLetterQueueName("shared-dlq"); + // Register a tenant that only supplies credentials, no region/endpoint of its own. + sqs.AddTenant("tenantB", new BasicAWSCredentials("k", "s")); + + var transport = options.Transports.GetOrCreate(); + transport.AutoProvision = true; + var tenant = transport.Tenants["tenantB"]; + tenant.Compile(transport, null!); + + tenant.Transport.Config.RegionEndpoint.ShouldBe(RegionEndpoint.USEast1); + tenant.Transport.AutoProvision.ShouldBeTrue(); + tenant.Transport.DefaultDeadLetterQueueName.ShouldBe("shared-dlq"); + } + + [Fact] + public void sqs_queues_are_tenant_aware_by_default() + { + var transport = new AmazonSqsTransport(); + var queue = transport.Queues["orders"]; + queue.TenancyBehavior.ShouldBe(TenancyBehavior.TenantAware); + } + + [Fact] + public void build_tenant_sibling_copies_configuration_but_uses_the_tenant_transport() + { + var transport = new AmazonSqsTransport(); + var tenant = transport.Tenants["tenantB"]; + + var queue = transport.Queues["orders"]; + queue.VisibilityTimeout = 42; + queue.MaxNumberOfMessages = 7; + queue.EnableFairQueueMessageGroups = true; + + var sibling = queue.BuildTenantSibling(tenant); + + sibling.ShouldNotBeSameAs(queue); + sibling.QueueName.ShouldBe("orders"); + sibling.VisibilityTimeout.ShouldBe(42); + sibling.MaxNumberOfMessages.ShouldBe(7); + sibling.EnableFairQueueMessageGroups.ShouldBeTrue(); + // Resolved from the tenant transport's own (isolated) queue cache. + tenant.Transport.Queues["orders"].ShouldBeSameAs(sibling); + } +} + +public record TenantColorMessage(string Color); + +public static class TenantColorMessageHandler +{ + public static void Handle(TenantColorMessage message) + { + // no-op; presence lets Wolverine discover a handler so receive tests can track processing + } +} diff --git a/src/Transports/AWS/Wolverine.AmazonSqs.Tests/AmazonSqsPerTenantConnectionTests.cs b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/AmazonSqsPerTenantConnectionTests.cs new file mode 100644 index 000000000..0136a7da5 --- /dev/null +++ b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/AmazonSqsPerTenantConnectionTests.cs @@ -0,0 +1,228 @@ +using Amazon.Runtime; +using Amazon.SQS; +using Amazon.SQS.Model; +using JasperFx.Core; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.Tracking; +using Wolverine.Transports.Sending; +using Xunit; + +namespace Wolverine.AmazonSqs.Tests; + +/// +/// Integration coverage for broker-per-tenant Amazon SQS (GH-3304). LocalStack is a single container and does not +/// enforce AWS account boundaries, so these tests cannot prove separate accounts. They instead simulate a +/// tenant with its own dedicated connection by pointing the tenant at a different signing region +/// (AuthenticationRegion) on the same LocalStack endpoint — LocalStack partitions SQS queues per region, so +/// the tenant's "colors" queue is genuinely a different physical queue from the shared/default one. That lets us +/// assert real message routing (a tenant message lands on the tenant queue and NOT the shared one, and vice versa) +/// plus inbound Envelope.TenantId stamping, rather than physical broker separation. +/// +/// Guarded to skip when LocalStack/Docker is unavailable. +/// +public class AmazonSqsPerTenantConnectionTests : IAsyncLifetime +{ + private const string ServiceUrl = "http://localhost:4566"; + private const string SharedRegion = "us-east-1"; + private const string TenantRegion = "us-west-2"; + + private bool _skip; + + public async Task InitializeAsync() + { + _skip = !await IsLocalStackAvailable(); + } + + public Task DisposeAsync() => Task.CompletedTask; + + [Fact] + public async Task tenant_message_is_published_to_the_tenant_region_and_not_the_default() + { + if (_skip) return; + + var queue = $"pertenant-{Guid.NewGuid():N}"; + using var host = await buildSenderAsync(queue); + + await host.MessageBus().SendAsync(new TenantColorMessage("for-tenant-b"), + new DeliveryOptions { TenantId = "tenantB" }); + + // Landed on the tenant region's queue... + (await consumeOne(TenantRegion, queue, 15.Seconds())).ShouldNotBeNull(); + // ...and NOT on the shared region's queue. + (await consumeOne(SharedRegion, queue, 3.Seconds())).ShouldBeNull(); + } + + [Fact] + public async Task default_message_is_published_to_the_shared_region() + { + if (_skip) return; + + var queue = $"pertenant-{Guid.NewGuid():N}"; + using var host = await buildSenderAsync(queue); + + await host.MessageBus().SendAsync(new TenantColorMessage("no-tenant")); + + // Falls back to the shared/default region (TenantedIdBehavior.FallbackToDefault)... + (await consumeOne(SharedRegion, queue, 15.Seconds())).ShouldNotBeNull(); + // ...and NOT the tenant region. + (await consumeOne(TenantRegion, queue, 3.Seconds())).ShouldBeNull(); + } + + [Fact] + public async Task tenant_message_is_consumed_and_stamped_with_the_tenant_id() + { + if (_skip) return; + + var queue = $"pertenant-{Guid.NewGuid():N}"; + + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.ServiceName = "PerTenantInbound"; + configureTransport(opts); + + opts.Policies.DisableConventionalLocalRouting(); + opts.PublishMessage().ToSqsQueue(queue).SendInline(); + opts.ListenToSqsQueue(queue); + }) + .StartAsync(); + + // The default listener polls the shared region and the tenant listener polls the tenant region; the message + // only exists in the tenant region, so only the tenant listener consumes it and stamps the tenant id. + var session = await host + .TrackActivity() + .IncludeExternalTransports() + .Timeout(60.Seconds()) + .WaitForMessageToBeReceivedAt(host) + .ExecuteAndWaitAsync(c => + c.SendAsync(new TenantColorMessage("for-tenant-b"), new DeliveryOptions { TenantId = "tenantB" })); + + var received = session.Received.SingleEnvelope(); + received.TenantId.ShouldBe("tenantB"); + received.Message.ShouldBeOfType().Color.ShouldBe("for-tenant-b"); + } + + [Fact] + public async Task tenant_aware_endpoint_resolves_a_TenantedSender() + { + if (_skip) return; + + var queue = $"pertenant-{Guid.NewGuid():N}"; + using var host = await buildSenderAsync(queue); + + var runtime = host.GetRuntime(); + var transport = runtime.Options.Transports.GetOrCreate(); + var endpoint = transport.Queues[queue]; + + // buildSenderAsync publishes inline, so the endpoint resolves an InlineSendingAgent around our sender. + var agent = (InlineSendingAgent)runtime.Endpoints.GetOrBuildSendingAgent(endpoint.Uri); + agent.Sender.ShouldBeOfType(); + + // Each tenant got its own compiled child transport with its own client. + transport.Tenants["tenantB"].Transport.Client.ShouldNotBeNull(); + } + + [Fact] + public async Task per_tenant_queues_are_provisioned_on_the_tenant_region() + { + if (_skip) return; + + var queue = $"pertenant-{Guid.NewGuid():N}"; + using var host = await buildSenderAsync(queue); + + // AutoProvision + ConnectAsync must have created the shared topology queue on the tenant's own region. + using var tenantClient = rawClient(TenantRegion); + var url = await tenantClient.GetQueueUrlAsync(queue); + url.QueueUrl.ShouldNotBeNull(); + url.QueueUrl.ShouldContain(TenantRegion); + } + + private static void configureTransport(WolverineOptions opts) + { + opts.UseAmazonSqsTransport(c => + { + c.ServiceURL = ServiceUrl; + c.AuthenticationRegion = SharedRegion; + }) + .Credentials(new BasicAWSCredentials("ignore", "ignore")) + .AutoProvision() + .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault) + // The tenant shares the LocalStack endpoint but signs for a different region, which LocalStack + // partitions into its own physical queue store — standing in for a dedicated tenant connection. + .AddTenant("tenantB", c => + { + c.ServiceURL = ServiceUrl; + c.AuthenticationRegion = TenantRegion; + }); + } + + private static Task buildSenderAsync(string queue) + { + return Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.ServiceName = "PerTenantSender"; + configureTransport(opts); + + opts.Policies.DisableConventionalLocalRouting(); + opts.PublishMessage().ToSqsQueue(queue).SendInline(); + }) + .StartAsync(); + } + + private static AmazonSQSClient rawClient(string region) + { + return new AmazonSQSClient(new BasicAWSCredentials("ignore", "ignore"), + new AmazonSQSConfig { ServiceURL = ServiceUrl, AuthenticationRegion = region }); + } + + // Read a single message from the given region/queue within the timeout, or null if none arrives. + private static async Task consumeOne(string region, string queueName, TimeSpan timeout) + { + using var client = rawClient(region); + + string queueUrl; + try + { + queueUrl = (await client.GetQueueUrlAsync(queueName)).QueueUrl; + } + catch (Amazon.SQS.Model.QueueDoesNotExistException) + { + return null; + } + + var deadline = DateTimeOffset.UtcNow.Add(timeout); + do + { + var response = await client.ReceiveMessageAsync(new ReceiveMessageRequest + { + QueueUrl = queueUrl, + MaxNumberOfMessages = 1, + WaitTimeSeconds = 1 + }); + + if (response.Messages is { Count: > 0 }) + { + return response.Messages[0]; + } + } while (DateTimeOffset.UtcNow < deadline); + + return null; + } + + private static async Task IsLocalStackAvailable() + { + try + { + using var client = rawClient(SharedRegion); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await client.ListQueuesAsync(new ListQueuesRequest(), cts.Token); + return true; + } + catch (Exception) + { + return false; + } + } +} diff --git a/src/Transports/AWS/Wolverine.AmazonSqs.Tests/Samples/Bootstrapping.cs b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/Samples/Bootstrapping.cs index 66068b46d..4a41824ed 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs.Tests/Samples/Bootstrapping.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/Samples/Bootstrapping.cs @@ -1,5 +1,6 @@ using System.Text; using System.Text.Json; +using Amazon; using Amazon.Runtime; using Amazon.SQS; using Amazon.SQS.Model; @@ -7,6 +8,7 @@ using JasperFx.Core; using Microsoft.Extensions.Hosting; using Wolverine.ComplianceTests.Compliance; +using Wolverine.Transports.Sending; namespace Wolverine.AmazonSqs.Tests.Samples; @@ -44,8 +46,51 @@ private async Task use_named_brokers() }).StartAsync(); #endregion - } - + } + + private async Task broker_per_tenant() + { + #region sample_sqs_broker_per_tenant + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + // The "default" / shared SQS connection + opts.UseAmazonSqsTransport(config => + { + config.RegionEndpoint = RegionEndpoint.USEast1; + }) + .AutoProvision() + + // How should Wolverine route a message whose TenantId is null or + // unknown? FallbackToDefault (the default) uses the shared connection; + // TenantIdRequired throws; IgnoreUnknownTenants silently drops it. + .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault) + + // Each tenant gets its OWN dedicated SQS connection, but shares the + // queue topology declared below. This tenant inherits the parent's + // AWS credentials, and just re-points at its own region. + .AddTenant("tenant-west", config => + { + config.RegionEndpoint = RegionEndpoint.USWest2; + }) + + // Or give the tenant its own dedicated AWS account by supplying + // its own credentials (optionally with its own region/endpoint too): + .AddTenant("tenant-eu", new BasicAWSCredentials("tenant-eu-key", "tenant-eu-secret"), + config => + { + config.RegionEndpoint = RegionEndpoint.EUWest1; + }); + + // One shared topology; messages are routed to the right connection at + // runtime by Envelope.TenantId (e.g. new DeliveryOptions { TenantId = "tenant-west" }). + opts.PublishMessage().ToSqsQueue("colors"); + opts.ListenToSqsQueue("colors"); + }).StartAsync(); + + #endregion + } + private async Task for_local_development() { #region sample_connect_to_sqs_and_localstack diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs index cbd0564f8..cbf05e494 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs @@ -383,21 +383,70 @@ public override async ValueTask BuildListenerAsync(IWolverineRuntime { throw new InvalidOperationException("The parent transport has not yet been initialized"); } - + Mapper ??= BuildMapper(runtime); + var logger = runtime.LoggerFactory.CreateLogger(); + if (QueueUrl.IsEmpty()) { - await InitializeAsync(runtime.LoggerFactory.CreateLogger()); + await InitializeAsync(logger); } - - return new SqsListener(runtime, this, _parent, receiver); + + var listener = new SqsListener(runtime, this, _parent, receiver); + + // Broker-per-tenant (GH-3304): the shared listener consumes the default account. Each tenant runs its own + // listener on its own account/region, stamping the tenant id onto inbound envelopes via TenantIdRule. + // Per-envelope completion routes back over the receiving connection through Envelope.Listener — the same + // CompoundListener multi-tenancy pattern used by RabbitMQ / NATS / Kafka. + if (_parent.Tenants.Any() && TenancyBehavior == TenancyBehavior.TenantAware) + { + var compound = new CompoundListener(Uri); + compound.Inner.Add(listener); + + foreach (var tenant in _parent.Tenants) + { + var tenantQueue = BuildTenantSibling(tenant); + if (tenantQueue.QueueUrl.IsEmpty()) + { + await tenantQueue.InitializeAsync(logger); + } + + var tenantReceiver = new ReceiverWithRules(receiver, [new TenantIdRule(tenant.TenantId)]); + compound.Inner.Add(new SqsListener(runtime, tenantQueue, tenant.Transport, tenantReceiver)); + } + + return compound; + } + + return listener; } protected override ISender CreateSender(IWolverineRuntime runtime) { Mapper ??= BuildMapper(runtime); - + + // Broker-per-tenant (GH-3304): route by Envelope.TenantId to a per-tenant sender bound to that tenant's own + // account, falling back to the shared account for the default/untenanted path. + // + // Both the tenant senders AND the default sender they fall back to must be simple fire-and-forget ISenders: + // TenantedSender intentionally does NOT implement ISenderRequiresCallback (GH-2361), and it does not forward + // RegisterCallback to the senders beneath it. A BatchedSender (SqsSenderProtocol) registered under it would + // therefore never receive its ISenderCallback and would silently drop every message. InlineSqsSender sends + // directly and needs no callback — the same fire-and-forget model the RabbitMQ / NATS / Kafka per-tenant + // senders use. + if (_parent.Tenants.Any() && TenancyBehavior == TenancyBehavior.TenantAware) + { + var tenantedSender = new TenantedSender(Uri, _parent.TenantedIdBehavior, new InlineSqsSender(runtime, this)); + foreach (var tenant in _parent.Tenants) + { + var tenantQueue = BuildTenantSibling(tenant); + tenantedSender.RegisterSender(tenant.TenantId, new InlineSqsSender(runtime, tenantQueue)); + } + + return tenantedSender; + } + if (Mode == EndpointMode.Inline) { return new InlineSqsSender(runtime, this); @@ -409,6 +458,51 @@ protected override ISender CreateSender(IWolverineRuntime runtime) runtime.LoggerFactory.CreateLogger()); } + /// + /// Broker-per-tenant (GH-3304): materialize this queue's tenant-specific twin on the given tenant's child + /// transport — same queue name and configuration, but bound to the tenant's own SQS client and its own + /// QueueUrl cache (which is why a fresh endpoint is required rather than reusing this one). The tenant twin is + /// cached on the tenant transport's so repeated sender/listener builds + /// resolve the same instance. + /// + internal AmazonSqsQueue BuildTenantSibling(AmazonSqsTenant tenant) + { + var sibling = tenant.Transport.Queues[QueueName]; + + sibling.Mode = Mode; + sibling.EndpointName = EndpointName; + sibling.IsListener = IsListener; + sibling.Role = Role; + sibling.EnableFairQueueMessageGroups = EnableFairQueueMessageGroups; + sibling.VisibilityTimeout = VisibilityTimeout; + sibling.WaitTimeSeconds = WaitTimeSeconds; + sibling.MaxNumberOfMessages = MaxNumberOfMessages; + sibling.MessageAttributeNames = MessageAttributeNames; + + // Share the interop mapper strategy so tenant traffic serializes identically to the shared account. + sibling.Mapper = Mapper; + sibling.MapperFactory = MapperFactory; + + // Preserve queue-creation attributes (FIFO, retention, redrive, ...) for AutoProvision on the tenant account. + if (Configuration.Attributes is { Count: > 0 }) + { + sibling.Configuration.Attributes ??= new Dictionary(); + foreach (var pair in Configuration.Attributes) + { + sibling.Configuration.Attributes[pair.Key] = pair.Value; + } + } + + // Only pin the dead letter queue name when it was set explicitly on this listener; otherwise let the tenant + // queue fall back to the tenant transport's own DefaultDeadLetterQueueName (seeded in AmazonSqsTenant.Compile). + if (_deadLetterQueueNameSetExplicitly) + { + sibling.DeadLetterQueueName = _deadLetterQueueName; + } + + return sibling; + } + protected override bool supportsMode(EndpointMode mode) { return true; diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTenant.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTenant.cs new file mode 100644 index 000000000..db62d1b64 --- /dev/null +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTenant.cs @@ -0,0 +1,86 @@ +using Amazon.SQS; +using JasperFx.Core; +using Wolverine.Runtime; + +namespace Wolverine.AmazonSqs.Internal; + +/// +/// Represents a single Wolverine tenant that is served by its own dedicated Amazon SQS connection (distinct AWS +/// account/credentials, region, or ServiceURL) while sharing the queue topology declared on the parent transport. +/// Because an SQS "connection" is credential + region config plus a lazily built +/// client, and because is cached per endpoint (and would collide across +/// tenants), each tenant owns a child whose config is seeded from the parent and +/// then re-pointed at the tenant's own account/region — giving the tenant its own client and its own queue +/// (and QueueUrl) cache. Outbound routing is by via the framework's +/// ; inbound the tenant's own listener stamps the tenant id. +/// Mirrors the Azure Service Bus tenant model. +/// +internal class AmazonSqsTenant +{ + public AmazonSqsTenant(string tenantId) + { + TenantId = tenantId ?? throw new ArgumentNullException(nameof(tenantId)); + Transport = new AmazonSqsTransport(); + } + + public string TenantId { get; } + + /// + /// The child transport that provides this tenant's dedicated SQS client and its own queue/QueueUrl cache. Its + /// connection config is seeded from the parent's and re-pointed at the tenant's own account/region during + /// . + /// + public AmazonSqsTransport Transport { get; } + + /// + /// Optional configuration hook applied to the tenant's own during + /// , after the parent's connection settings have been seeded onto it — so the + /// tenant only overrides the axes it actually sets (typically region / ServiceURL / AuthenticationRegion). Runs + /// at compile time rather than registration time because the parent connection is not fully configured until + /// bootstrap completes. Null when the tenant is configured purely by supplying its own credentials. + /// + public Action? Configure { get; set; } + + /// + /// Seed the tenant's child transport from the parent (credentials + connection endpoint + provisioning / DLQ + /// behavior), apply the tenant's own overrides, then build the tenant's SQS client. + /// Called from once the parent connection has been fully resolved. + /// The seed-then-override order (mirroring the Kafka tenant model) is what lets a tenant re-point a single axis — + /// e.g. just its region — while inheriting everything else from the parent. Note that + /// is never truly null (it lazily resolves to an ambient default), + /// so inheritance cannot rely on a null check — it seeds unconditionally and lets the tenant action win. + /// + public AmazonSqsTransport Compile(AmazonSqsTransport parent, IWolverineRuntime runtime) + { + // Inherit the parent credential source unless the tenant supplied its own (dedicated-account case). + Transport.CredentialSource ??= parent.CredentialSource; + + // Seed the parent's connection endpoint. ServiceURL and RegionEndpoint are mutually exclusive on + // AmazonSQSConfig (setting one clears the other), so copy whichever the parent is actually using; + // AuthenticationRegion is independent (it lets a shared endpoint like LocalStack sign for a distinct region). + if (parent.Config.ServiceURL.IsNotEmpty()) + { + Transport.Config.ServiceURL = parent.Config.ServiceURL; + } + else if (parent.Config.RegionEndpoint != null) + { + Transport.Config.RegionEndpoint = parent.Config.RegionEndpoint; + } + + Transport.Config.AuthenticationRegion = parent.Config.AuthenticationRegion; + + // The tenant's own overrides win over the seeded parent settings. + Configure?.Invoke(Transport.Config); + + // Provision + dead-letter behavior must match the parent so tenant queues are created and dead-lettered the + // same way on the tenant's own account. + Transport.AutoProvision = parent.AutoProvision; + Transport.AutoPurgeAllQueues = parent.AutoPurgeAllQueues; + Transport.DisableDeadLetterQueues = parent.DisableDeadLetterQueues; + Transport.DefaultDeadLetterQueueName = parent.DefaultDeadLetterQueueName; + + Transport.Client ??= Transport.BuildClient(runtime); + + return Transport; + } +} diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransport.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransport.cs index 8e1d188d4..aaa1d58de 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransport.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransport.cs @@ -11,7 +11,7 @@ namespace Wolverine.AmazonSqs.Internal; -public class AmazonSqsTransport : BrokerTransport +public class AmazonSqsTransport : BrokerTransport, IAsyncDisposable { public const string DeadLetterQueueName = DeadLetterQueueConstants.DefaultQueueName; public const string ResponseEndpointName = "AmazonSqsResponses"; @@ -95,10 +95,18 @@ internal AmazonSqsTransport(IAmazonSQS client) : this() [IgnoreDescription] public LightweightCache Queues { get; } + /// + /// Broker-per-tenant registrations (GH-3304). Each tenant owns a child transport pointed at its own SQS + /// account/region/endpoint (and its own queue + QueueUrl cache); outbound is routed by + /// and inbound listeners stamp the tenant id. + /// + [IgnoreDescription] + internal LightweightCache Tenants { get; } = new(name => new AmazonSqsTenant(name)); + [ChildDescription] public AmazonSQSConfig Config { get; } = new(); - internal IAmazonSQS? Client { get; private set; } + internal IAmazonSQS? Client { get; set; } public int LocalStackPort { get; set; } @@ -219,6 +227,59 @@ public override async ValueTask ConnectAsync(IWolverineRuntime runtime) await CleanupOrphanedSystemQueuesAsync(runtime); StartSystemQueueKeepAlive(runtime.DurabilitySettings.Cancellation, runtime); } + + // Broker-per-tenant (GH-3304): now that the parent connection is fully resolved, seed each tenant's child + // transport from it and build the tenant's own SQS client. Clients are lightweight and the queues/QueueUrls + // are resolved lazily per tenant, so there's no live connection to open here — but when AutoProvision is on + // we create the shared queue topology on every tenant account, since each is a separate broker. + if (Tenants.Any()) + { + foreach (var tenant in Tenants) + { + tenant.Compile(this, runtime); + } + + if (AutoProvision) + { + await provisionTenantQueuesAsync(runtime); + } + } + } + + // Create the shared application queues (and their dead letter queues) on each tenant's own account. Mirrors + // what the default connection provisions lazily via AmazonSqsQueue.InitializeAsync, but runs against the + // tenant client so a tenant's listener has its queue ready before it starts polling (GH-3304). + private async Task provisionTenantQueuesAsync(IWolverineRuntime runtime) + { + var logger = runtime.LoggerFactory.CreateLogger(); + + var applicationQueues = Queues + .Where(x => x.Role == EndpointRole.Application) + .ToArray(); + + foreach (var tenant in Tenants) + { + foreach (var queue in applicationQueues) + { + try + { + var tenantQueue = queue.BuildTenantSibling(tenant); + await tenantQueue.SetupAsync(tenant.Transport.Client!); + + if (tenantQueue.DeadLetterQueueName.IsNotEmpty() && !DisableDeadLetterQueues) + { + await tenant.Transport.Queues[tenantQueue.DeadLetterQueueName!] + .SetupAsync(tenant.Transport.Client!); + } + } + catch (Exception e) + { + logger.LogWarning(e, + "Error while provisioning queue {Queue} for tenant {TenantId}", queue.QueueName, + tenant.TenantId); + } + } + } } public WolverineTransportHealthCheck BuildHealthCheck(IWolverineRuntime runtime) @@ -347,4 +408,17 @@ internal void ConnectToLocalStack(int port = 4566) } public string ServerHost => Config.ServiceURL?.ToUri().Host!; + + public ValueTask DisposeAsync() + { + Client?.Dispose(); + + // Broker-per-tenant (GH-3304): each tenant owns its own SQS client through its child transport; dispose them. + foreach (var tenant in Tenants) + { + tenant.Transport.Client?.Dispose(); + } + + return ValueTask.CompletedTask; + } } \ No newline at end of file diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransportConfiguration.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransportConfiguration.cs index 9b8dffe4d..d6c5abba8 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransportConfiguration.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsTransportConfiguration.cs @@ -1,8 +1,11 @@ using Amazon.Runtime; +using Amazon.SQS; +using JasperFx.Core; using Microsoft.Extensions.DependencyInjection; using Wolverine.Configuration; using Wolverine.Runtime; using Wolverine.Transports; +using Wolverine.Transports.Sending; namespace Wolverine.AmazonSqs.Internal; @@ -266,4 +269,66 @@ public AmazonSqsTransportConfiguration EnableWolverineControlQueues() return this; } + + /// + /// Override the sending behavior for unknown or missing tenant ids when using broker-per-tenant Amazon SQS + /// multi-tenancy (GH-3304). See . Default is + /// unless changed. + /// + /// + /// + public AmazonSqsTransportConfiguration TenantIdBehavior(TenantedIdBehavior behavior) + { + Transport.TenantedIdBehavior = behavior; + return this; + } + + /// + /// Register a tenant that is served by its own dedicated Amazon SQS connection (typically a distinct region or + /// ServiceURL) while sharing the queue topology declared on this transport. The tenant inherits the + /// parent's AWS credentials and provisioning / dead-letter behavior; use to point + /// the tenant at its own region or endpoint. Outbound messages carrying a matching + /// are routed to this tenant's connection; inbound messages consumed from it are + /// stamped with the tenant id. + /// + /// + /// Configuration applied to the tenant's own . + /// + public AmazonSqsTransportConfiguration AddTenant(string tenantId, Action configure) + { + if (tenantId.IsEmpty()) throw new ArgumentOutOfRangeException(nameof(tenantId), "Empty or null tenantId"); + ArgumentNullException.ThrowIfNull(configure); + + // Deferred: applied in AmazonSqsTenant.Compile() after the parent connection is seeded onto the tenant, so + // the tenant only overrides the axes it sets and inherits the rest. + Transport.Tenants[tenantId].Configure = configure; + + return this; + } + + /// + /// Register a tenant that is served by its own dedicated Amazon SQS account, identified by its own + /// , while sharing the queue topology declared on this transport. Use the optional + /// to also point the tenant at its own region or ServiceURL; if omitted the + /// tenant inherits the parent's region/endpoint. Outbound messages carrying a matching + /// are routed to this tenant's account; inbound messages consumed from it are + /// stamped with the tenant id. + /// + /// + /// The AWS credentials for the tenant's dedicated account. + /// Optional configuration applied to the tenant's own . + /// + public AmazonSqsTransportConfiguration AddTenant(string tenantId, AWSCredentials credentials, + Action? configure = null) + { + if (tenantId.IsEmpty()) throw new ArgumentOutOfRangeException(nameof(tenantId), "Empty or null tenantId"); + ArgumentNullException.ThrowIfNull(credentials); + + var tenant = Transport.Tenants[tenantId]; + tenant.Transport.CredentialSource = _ => credentials; + // Deferred: applied in AmazonSqsTenant.Compile() after the parent connection is seeded onto the tenant. + tenant.Configure = configure; + + return this; + } } \ No newline at end of file