NATS transport: dynamic subjects, JetStream dedup, per-tenant connection

Directory.Packages.props

@@ -70,7 +70,7 @@
     <PackageVersion Include="Microsoft.VisualStudio.Threading.Analyzers" Version="17.0.64" />
     <PackageVersion Include="MQTTnet.Extensions.ManagedClient" Version="4.3.7.1207" />
     <PackageVersion Include="MySqlConnector" Version="2.4.0" />
-    <PackageVersion Include="NATS.Net" Version="2.7.0" />
+    <PackageVersion Include="NATS.Net" Version="2.8.2" />
     <PackageVersion Include="NewId" Version="4.0.1" />
     <PackageVersion Include="Newtonsoft.Json" Version="13.0.3" />
     <PackageVersion Include="Npgsql" Version="9.0.3" />

docs/guide/messaging/transports/nats.md

@@ -299,6 +299,24 @@ opts.ListenToNatsSubject("orders.received")
     .Named("orders-listener");
 ```
 
+### Load Balancing with Queue Groups
+
+Multiple listeners sharing a NATS queue group have each message delivered to only one member, spreading load
+across instances. Set a transport-wide default so every listener joins the same group:
+
+```csharp
+opts.UseNats(nats =>
+{
+    nats.ConnectionString = "nats://localhost:4222";
+    nats.DefaultQueueGroup = "orders-workers";
+});
+```
+
+::: tip Subject normalization
+By default the transport normalizes `/` separators in subjects to NATS `.` tokens (`NormalizeSubjects`, on by
+default). Set `nats.NormalizeSubjects = false` if you need to use literal subjects that contain `/`.
+:::
+
 ## Publishing Messages
 
 ### To a Specific Subject
@@ -325,6 +343,88 @@ opts.PublishAllMessages()
     .SendInline();
 ```
 
+### Static Outgoing Headers
+
+Attach a constant header to every message published to a subject with `AddOutgoingHeader`:
+
+```csharp
+opts.PublishMessage<OrderCreated>()
+    .ToNatsSubject("orders.created")
+    .AddOutgoingHeader("x-source", "orders-service");
+```
+
+### Per-Message (Dynamic) Subjects
+
+`ToNatsSubject("...")` publishes to a single static subject. To compute the subject *per message* — e.g. an
+aggregate-scoped subject like `orders.events.{id}` — use `PublishMessagesToNatsSubject<T>`. This is built on
+Wolverine's generic topic routing (`RoutingMode.ByTopic` / `Envelope.TopicName`), the same mechanism the
+RabbitMQ, Kafka, and MQTT transports use, so it also participates in `IMessageBus.BroadcastToTopicAsync`.
+
+```csharp
+opts.UseNats("nats://localhost:4222").AutoProvision();
+
+// The subject is derived from each message instance.
+opts.PublishMessagesToNatsSubject<OrderEvent>(e => $"orders.events.{e.OrderId}");
+```
+
+The same endpoint is automatically enrolled for explicit topic broadcasts, where the caller supplies the
+subject directly (overriding the function):
+
+```csharp
+await bus.BroadcastToTopicAsync("orders.events.12345", new OrderShipped(...));
+```
+
+::: tip Consuming dynamic subjects
+Because the publish subject varies, a consumer must subscribe to the whole space with a NATS wildcard.
+For Core NATS, listen on `orders.events.>`. For JetStream, provision the stream over a wildcard subject
+(`orders.events.>`) so it captures every computed subject, then listen with a matching consumer filter. A
+too-narrow stream subject silently fails to capture the dynamic subjects.
+:::
+
+For subject shaping that a strongly-typed `Func<T, string>` can't express — for example deriving the subject
+from an envelope header or tenant id — configure an `ISubjectResolver`. It runs after the base/topic subject
+is determined and can rewrite it from any envelope state:
+
+```csharp
+opts.UseNats(nats =>
+{
+    nats.ConnectionString = "nats://localhost:4222";
+    nats.SubjectResolver = new MyAggregateSubjectResolver();
+});
+```
+
+### Deduplication (JetStream `Nats-Msg-Id`)
+
+Wolverine stamps a `Nats-Msg-Id` on every JetStream publish, so the stream's duplicate window discards
+duplicates server-side — the idempotency key external (non-Wolverine) consumers can rely on, independent of
+Wolverine's own durable-inbox dedup on `Envelope.Id`.
+
+By default the id is the Wolverine `Envelope.Id`. Project a domain identity instead with `DeduplicateUsing`
+so a logical event dedups even across separate sends (e.g. `{stream}/{version}`):
+
+```csharp
+opts.UseNats("nats://localhost:4222")
+    .AutoProvision()
+    // Any two publishes resolving to the same key within the stream's duplicate window collapse to one.
+    .DeduplicateUsing(envelope => $"{envelope.GroupId}/{envelope.Id}");
+```
+
+Precedence for the dedup key:
+
+1. An explicit `Nats-Msg-Id` header already on the outgoing envelope always wins.
+2. Otherwise the configured `DeduplicateUsing` function is used.
+3. Otherwise the Wolverine `Envelope.Id`.
+
+The duplicate window itself is configured per stream (`WithDeduplicationWindow`) or transport-wide via
+`JetStreamDefaults.DuplicateWindow` (default two minutes):
+
+```csharp
+opts.UseNats("nats://localhost:4222")
+    .DefineStream("ORDERS", s => s
+        .WithSubjects("orders.>")
+        .WithDeduplicationWindow(TimeSpan.FromMinutes(5)));
+```
+
 ## Scheduled Message Delivery
 
 NATS Server 2.12+ supports native scheduled message delivery. When enabled, Wolverine uses NATS headers for scheduling instead of database persistence.
@@ -380,9 +480,26 @@ When native scheduled send is not available (server < 2.12 or stream not configu
 
 ## Multi-Tenancy
 
-NATS transport supports subject-based tenant isolation.
+::: tip
+For a holistic overview of multi-tenancy across all of Wolverine, see the [Multi-Tenancy Tutorial](/tutorials/multi-tenancy)
+and [Multi-Tenancy with Wolverine](/guide/handlers/multi-tenancy) for how Wolverine tracks the tenant id across messages.
+:::
+
+The NATS transport supports two flavors of tenant isolation:
+
+- **Subject-based** — all tenants share one connection and are separated by a tenant subject prefix
+  (`{tenantId}.{subject}`). This is soft partitioning within a single NATS account.
+- **Connection-based** — a tenant gets its own dedicated NATS connection to a different server or **account**.
 
-### Basic Multi-Tenancy
+::: info NATS accounts are the native tenancy boundary
+In NATS, true multi-tenancy is [Accounts](https://docs.nats.io/running-a-nats-service/configuration/securing_nats/accounts):
+each account is a fully isolated subject namespace, and a single connection authenticates into exactly **one**
+account. So a genuinely isolated tenant means a **dedicated connection with its own credentials** (see
+[Per-Tenant Connections](#per-tenant-connections)). A subject prefix on a shared connection is only
+partitioning within one account, not account-level isolation.
+:::
+
+### Basic Multi-Tenancy (Subject Isolation)
 
 ```csharp
 opts.UseNats("nats://localhost:4222")
@@ -396,6 +513,33 @@ opts.UseNats("nats://localhost:4222")
 - `TenantIdRequired`: Throws if tenant ID is missing
 - `FallbackToDefault`: Uses base subject if tenant ID is missing
 
+### Per-Tenant Connections
+
+To route a tenant to its own NATS server or account, add it with a configuration action. The action receives
+a copy of the transport's own connection settings, so you only override what differs for this tenant — a
+different URL, or any of the NATS auth mechanisms (token, JWT/NKey, credentials file, client certificate):
+
+```csharp
+opts.UseNats("nats://shared:4222")
+    .ConfigureMultiTenancy(TenantedIdBehavior.FallbackToDefault)
+    .AddTenant("tenant-a", cfg => cfg.ConnectionString = "nats://tenant-a-host:4222")
+    .AddTenant("tenant-b", cfg =>
+    {
+        cfg.ConnectionString = "nats://tenant-b-host:4222";
+        cfg.CredentialsFile = "/etc/nats/tenant-b.creds";
+    });
+```
+
+Each tenant with its own configuration gets a dedicated connection, owned by the transport for its lifetime.
+Tenants added without a configuration action keep sharing the transport connection (subject-prefix isolation
+only).
+
+Both **sending and listening** are tenant-aware. A listener consumes on the shared connection *and* on each
+tenant's dedicated connection: when a message arrives on a tenant connection it is stamped with that tenant's
+id, and its ack/nak/dead-letter is routed back over the same connection. Sending a message tagged with a
+`TenantId` publishes it over that tenant's connection. If any tenant streams need JetStream, the configured
+streams are auto-provisioned on each tenant server as well (when `AutoProvision()` is on).
+
 ### Custom Subject Mapper
 
 ```csharp
@@ -430,8 +574,13 @@ The response endpoint always uses Core NATS for low-latency replies, even when t
 
 ### JetStream
 
-- **Retry**: Message is requeued via `NakAsync()` with optional delay
-- **Dead Letter**: Message is terminated via `AckTerminateAsync()`
+- **Retry**: Message is requeued via `NakAsync()` with optional delay, up to the consumer's maximum delivery
+  attempts (`JetStreamDefaults.MaxDeliver`, default 5, or a per-endpoint `MaxDeliveryAttempts` override).
+- **Dead Letter**: Once delivery attempts are exhausted, the poison message is first forwarded to the
+  configured dead-letter subject (so a terminate failure can't lose it), then terminated on the consumer via
+  `AckTerminateAsync(reason)` so the server stops redelivering and records why. If **no** dead-letter subject
+  is configured, Wolverine logs a warning and the message is terminated without being retained — configure a
+  dead-letter subject to keep poison messages.
 
 ### Core NATS
 

src/Transports/NATS/Wolverine.Nats.Tests/Helpers/NatsTestHelpers.cs

@@ -0,0 +1,106 @@
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using NATS.Client.Core;
+
+namespace Wolverine.Nats.Tests;
+
+/// <summary>
+/// Shared helpers for the NATS integration tests. The whole NATS suite resolves the broker from the
+/// NATS_URL environment variable (set by <see cref="NatsContainerFixture"/> when Testcontainers is used),
+/// falling back to the docker-compose broker on localhost:4222.
+/// </summary>
+internal static class NatsTestHelpers
+{
+    public static string ResolveUrl()
+    {
+        return Environment.GetEnvironmentVariable("NATS_URL") ?? "nats://localhost:4222";
+    }
+
+    /// <summary>
+    /// Probe the broker by spinning up a throwaway Wolverine host. Returns false (so the caller can skip)
+    /// when no NATS server is reachable, mirroring the guard used across the existing NATS integration tests.
+    /// </summary>
+    public static async Task<bool> IsNatsAvailable(string natsUrl)
+    {
+        try
+        {
+            using var testHost = await Host.CreateDefaultBuilder()
+                .UseWolverine(opts => opts.UseNats(natsUrl))
+                .StartAsync();
+
+            await testHost.StopAsync();
+            return true;
+        }
+        catch
+        {
+            return false;
+        }
+    }
+
+    /// <summary>
+    /// Subscribe a raw NATS client to an exact subject. <c>SubscribeCoreAsync</c> registers the subscription
+    /// synchronously (the SUB is written before it returns) and the follow-up ping flushes it to the server,
+    /// so there is no subscribe-before-publish race. Used to prove the exact concrete subject a message was
+    /// published to — something a wildcard Wolverine listener can't (the receiving pipeline overwrites
+    /// <c>Envelope.Destination</c> with the listener's own address).
+    /// </summary>
+    public static async Task<RawSubscription> SubscribeRawAsync(string url, string subject)
+    {
+        var connection = new NatsConnection(new NatsOpts { Url = url });
+        await connection.ConnectAsync();
+        var sub = await connection.SubscribeCoreAsync<byte[]>(subject);
+        await connection.PingAsync();
+        return new RawSubscription(connection, sub);
+    }
+}
+
+/// <summary>
+/// A raw NATS subscription plus its dedicated connection, disposed together.
+/// </summary>
+internal sealed class RawSubscription : IAsyncDisposable
+{
+    private readonly NatsConnection _connection;
+    private readonly INatsSub<byte[]> _sub;
+
+    public RawSubscription(NatsConnection connection, INatsSub<byte[]> sub)
+    {
+        _connection = connection;
+        _sub = sub;
+    }
+
+    /// <summary>
+    /// Read the next message, returning null if none arrives within <paramref name="timeout"/>.
+    /// </summary>
+    public async Task<NatsMsg<byte[]>?> ReadAsync(TimeSpan timeout)
+    {
+        using var cts = new CancellationTokenSource(timeout);
+        try
+        {
+            return await _sub.Msgs.ReadAsync(cts.Token);
+        }
+        catch (OperationCanceledException)
+        {
+            return null;
+        }
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _sub.DisposeAsync();
+        await _connection.DisposeAsync();
+    }
+}
+
+/// <summary>
+/// Message used by the dynamic-subject and dedup tests. The <see cref="OrderId"/> doubles as a stable
+/// domain identity for JetStream deduplication (see <c>DeduplicateUsing</c>).
+/// </summary>
+public record OrderPlaced(string OrderId);
+
+public class OrderPlacedHandler
+{
+    // No-op: receivers only need this so Wolverine tracking counts the message as "received".
+    public void Handle(OrderPlaced message)
+    {
+    }
+}

src/Transports/NATS/Wolverine.Nats.Tests/MultiTenancyTests.cs

@@ -260,10 +260,11 @@ public class NatsTenantTests
     public void creates_tenant_with_id()
     {
         var tenant = new NatsTenant("tenant1");
-        
+
         tenant.TenantId.ShouldBe("tenant1");
         tenant.SubjectMapper.ShouldBeNull();
-        tenant.ConnectionString.ShouldBeNull();
+        tenant.ConnectionConfiguration.ShouldBeNull();
+        tenant.HasOwnConnection.ShouldBeFalse();
     }
 
     [Fact]

src/Transports/NATS/Wolverine.Nats.Tests/NatsDynamicSubjectTenancyTests.cs

@@ -0,0 +1,76 @@
+using IntegrationTests;
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine.Transports.Sending;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Wolverine.Nats.Tests;
+
+/// <summary>
+/// Regression coverage for the intersection of the two features this branch adds: per-message dynamic subjects
+/// (<c>RoutingMode.ByTopic</c> / <c>Envelope.TopicName</c>) and subject-isolation multi-tenancy.
+///
+/// A static endpoint subject is tenant-qualified once, at sender construction (<c>NatsEndpoint.CreateSender</c>).
+/// A subject computed per message is not — so without the fix a subject-isolation tenant's dynamic-subject send
+/// would publish to the raw, un-prefixed subject on the shared connection, silently defeating isolation. This
+/// asserts the computed subject is tenant-qualified (<c>{tenantId}.{computed}</c>) and does not leak to the other
+/// tenant's subject or the bare (un-prefixed) subject.
+///
+/// Uses raw NATS subscribers on the exact expected subjects (a single shared broker), so it is a local sanity
+/// check and is <b>not</b> required to run in CI.
+/// </summary>
+[Collection("NATS Integration")]
+[Trait("Category", "Integration")]
+public class NatsDynamicSubjectTenancyTests
+{
+    private readonly ITestOutputHelper _output;
+
+    public NatsDynamicSubjectTenancyTests(ITestOutputHelper output) => _output = output;
+
+    [Fact]
+    public async Task dynamic_subject_is_tenant_qualified_for_subject_isolation_tenants()
+    {
+        var url = NatsTestHelpers.ResolveUrl();
+        if (!await NatsTestHelpers.IsNatsAvailable(url))
+        {
+            return;
+        }
+
+        var root = $"orders.events.{Guid.NewGuid():N}";
+        var orderId = Guid.NewGuid().ToString("N");
+
+        using var host = await Host.CreateDefaultBuilder()
+            .ConfigureLogging(l => l.AddXunitLogging(_output))
+            .UseWolverine(opts =>
+            {
+                opts.ServiceName = "DynamicSubjectTenancy";
+                opts.UseNats(url)
+                    .ConfigureMultiTenancy(TenantedIdBehavior.FallbackToDefault)
+                    // Both tenants share the connection; isolation is purely by subject prefix.
+                    .AddTenant("tenantA")
+                    .AddTenant("tenantB");
+
+                opts.Policies.DisableConventionalLocalRouting();
+                opts.PublishMessagesToNatsSubject<OrderPlaced>(m => $"{root}.{m.OrderId}").SendInline();
+            })
+            .StartAsync();
+
+        // Each tenant's dynamic subject must be tenant-qualified: {tenantId}.{computed subject}.
+        await using var subA = await NatsTestHelpers.SubscribeRawAsync(url, $"tenantA.{root}.{orderId}");
+        await using var subB = await NatsTestHelpers.SubscribeRawAsync(url, $"tenantB.{root}.{orderId}");
+        // The un-prefixed subject proves nothing escapes isolation (this is where the pre-fix bug landed).
+        await using var subBare = await NatsTestHelpers.SubscribeRawAsync(url, $"{root}.{orderId}");
+
+        await host.MessageBus().SendAsync(new OrderPlaced(orderId), new DeliveryOptions { TenantId = "tenantA" });
+
+        var onA = await subA.ReadAsync(15.Seconds());
+        onA.ShouldNotBeNull();
+        onA!.Value.Subject.ShouldBe($"tenantA.{root}.{orderId}");
+
+        // Not on the other tenant's subject, and not on the bare un-prefixed subject.
+        (await subB.ReadAsync(2.Seconds())).ShouldBeNull();
+        (await subBare.ReadAsync(2.Seconds())).ShouldBeNull();
+    }
+}