NATS transport: dynamic subjects, JetStream dedup, per-tenant connection#3283
Merged
jeremydmiller merged 4 commits intoJul 5, 2026
Merged
Conversation
…ions; upgrade to NATS.Net 2.8.2 Audit and hardening of the NATS transport for per-aggregate routing and server-side idempotency. - Upgrade NATS.Net 2.7.0 -> 2.8.2 (no source changes required). - Per-message dynamic subjects via Wolverine's generic topic routing: PublishMessagesToNatsSubject<T>(Func<T,string>), IMessageBus.BroadcastToTopicAsync, and a wired ISubjectResolver escape hatch. NatsSender honors Envelope.TopicName, mirroring RabbitMqSender / InlineKafkaSender. - Overridable JetStream Nats-Msg-Id so the stream duplicate window actually deduplicates for external (non-Wolverine) consumers: DeduplicateUsing(Func<Envelope,string>), explicit Nats-Msg-Id header precedence, and wired DuplicateWindow provisioning. - Multi-tenancy: per-tenant dedicated connections with the full auth/TLS surface and tenant-owned NatsConnection; per-tenant inbound listening via a CompoundListener (one listener per tenant connection, tenant id stamped), mirroring the RabbitMQ / Azure Service Bus pattern. Configured streams are auto-provisioned on each tenant server. - Real dead-letter terminate: forward to the dead-letter subject, then AckTerminateAsync(reason); warn when no dead-letter subject is configured. - Wire previously-dead config: CustomHeaders (AddOutgoingHeader), DefaultQueueGroup, NormalizeSubjects, the JetStreamDefaults template, and JetStream domain / api-prefix through a centralized JetStream-context factory. - Delete dead code (ToJetStreamOpts, TenantAwareNatsSender, NatsEndpoint.BuildHeaders) and collapse divergent stream provisioning onto StreamConfiguration / JetStreamDefaults. - Tests: dynamic subjects, BroadcastToTopicAsync, subject resolver, JetStream dedup, and per-tenant send + inbound (two-broker). Docs updated, including NATS account-based multi-tenancy semantics.
There was a problem hiding this comment.
Pull request overview
This PR hardens and extends Wolverine’s NATS transport to support dynamic per-message subjects (topic routing), JetStream server-side deduplication via Nats-Msg-Id, and multi-tenancy with per-tenant dedicated NATS connections (including inbound listeners), alongside updated docs/tests and a NATS.Net client upgrade.
Changes:
- Add dynamic subject publishing via Wolverine topic routing (
RoutingMode.ByTopic/Envelope.TopicName) plus anISubjectResolverescape hatch. - Implement JetStream dedup key projection (
DeduplicateUsing) and apply stream duplicate windows during provisioning. - Add per-tenant dedicated connections and tenant-aware inbound listening via a compound listener pattern; remove superseded/dead code and wire previously inert config.
Reviewed changes
Copilot reviewed 21 out of 21 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| src/Transports/NATS/Wolverine.Nats/Internal/TenantAwareNatsSender.cs | Removed superseded tenant sender wrapper. |
| src/Transports/NATS/Wolverine.Nats/Internal/NatsTransportExpression.cs | Adds DeduplicateUsing and new tenant connection overloads (seeded from transport config). |
| src/Transports/NATS/Wolverine.Nats/Internal/NatsTransport.cs | Centralizes JetStream context creation, per-tenant connections, stream provisioning dedup window, and cleanup. |
| src/Transports/NATS/Wolverine.Nats/Internal/NatsTenant.cs | Replaces credential fields with per-tenant NatsTransportConfiguration and owned connection. |
| src/Transports/NATS/Wolverine.Nats/Internal/NatsSender.cs | Implements per-message subject routing + resolver hook; passes JetStream context into publishers. |
| src/Transports/NATS/Wolverine.Nats/Internal/NatsListener.cs | Implements “forward then terminate” dead-lettering with warning when DLQ subject missing; JetStream context injection. |
| src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs | Adds subject normalization toggle usage, default queue group behavior, per-tenant connection send/listen support, and JetStream defaults plumbing. |
| src/Transports/NATS/Wolverine.Nats/Internal/JetStreamSubscriber.cs | Uses injected JetStream context; respects effective queue group and JetStream defaults. |
| src/Transports/NATS/Wolverine.Nats/Internal/JetStreamPublisher.cs | Adds dedup key handling (Nats-Msg-Id) with precedence rules; uses publish opts for dedup. |
| src/Transports/NATS/Wolverine.Nats/Internal/CoreNatsSubscriber.cs | Uses effective queue group (endpoint override or transport default). |
| src/Transports/NATS/Wolverine.Nats/Extensions/NatsTransportExtensions.cs | Adds PublishMessagesToNatsSubject<T>(Func<T,string>) using topic routing. |
| src/Transports/NATS/Wolverine.Nats/Configuration/StreamConfiguration.cs | Adds per-stream DuplicateWindow + fluent WithDeduplicationWindow. |
| src/Transports/NATS/Wolverine.Nats/Configuration/NatsTransportConfiguration.cs | Adds MsgIdSource and documents JetStream defaults usage. |
| src/Transports/NATS/Wolverine.Nats/Configuration/NatsSubscriberConfiguration.cs | Adds AddOutgoingHeader for static per-endpoint outgoing headers. |
| src/Transports/NATS/Wolverine.Nats.Tests/NatsPerTenantConnectionTests.cs | New integration coverage for dedicated per-tenant connections (send + inbound). |
| src/Transports/NATS/Wolverine.Nats.Tests/NatsJetStreamDedupTests.cs | New integration coverage for JetStream server-side dedup precedence and behavior. |
| src/Transports/NATS/Wolverine.Nats.Tests/NatsDynamicSubjectTests.cs | New integration coverage for dynamic subjects, broadcasts-to-topic, and ISubjectResolver. |
| src/Transports/NATS/Wolverine.Nats.Tests/MultiTenancyTests.cs | Updates assertions for new tenant config representation. |
| src/Transports/NATS/Wolverine.Nats.Tests/Helpers/NatsTestHelpers.cs | Adds shared helpers for NATS integration tests + test message/handler. |
| docs/guide/messaging/transports/nats.md | Updates docs for dynamic subjects, dedup, queue groups, normalization, and connection-based multi-tenancy. |
| Directory.Packages.props | Upgrades NATS.Net from 2.7.0 to 2.8.2. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Contributor
Author
|
Seems we need to address this issue in a separate branch: I guess we are waiting on this: dotnet/aspnetcore#67464 without some hacky target file change to install and override dependencies to latest Microsoft.OpenApi pacakge, he suggests: dotnet/aspnetcore#67505 (comment) |
… doc - cloneConnectionConfiguration() now deep-copies the mutable JetStreamDefaults and Streams members. The reflective property copy aliased them by reference, so a tenant action that mutated them in place (e.g. cfg.JetStreamDefaults.X = ... or cfg.Streams[...] = ...) would leak into the shared transport config and every other tenant. Each tenant now gets its own copy. - Rewrite the NatsPerTenantConnectionTests class doc: it still claimed inbound over a tenant connection was a "not-yet-wired limitation," but the suite now includes a test proving inbound consumption round-trips over the tenant's own connection stamped with its tenant id.
Dynamic subjects (RoutingMode.ByTopic / Envelope.TopicName) took the publish subject straight from TopicName, bypassing the tenant subject mapping that subject-isolation tenancy relies on. A static endpoint subject is tenant- qualified once at sender construction (NatsEndpoint.CreateSender), but a per-message computed subject was not — so a subject-isolation tenant's ByTopic sends published to the raw, un-prefixed subject on the shared connection, silently defeating isolation (two tenants collide on the same subject). The per-tenant NatsSender now carries its tenant id + subject mapper and applies the same MapSubject to the computed subject, so static and dynamic sends isolate identically. Adds a regression test proving the computed subject is tenant- qualified and leaks to neither the other tenant nor the bare subject.
…isioning Addresses two more PR review comments: - SubjectResolver output now runs through NormalizeSubject, so a resolver that returns e.g. a '/'-separated subject is published with '.' tokens when NormalizeSubjects is on — consistent with static subjects and TopicName routing (a no-op beyond trimming when the flag is off). Adds a test that fails without the normalization. - The claim that dedicated tenant connections "will fail at runtime" because they are never explicitly ConnectAsync'd is a false positive: the NATS client connects lazily on first use, and send/consume over a tenant connection is already covered by passing tests. Adds a test proving JetStream stream auto-provisioning over a tenant's own connection succeeds and the stream exists on the tenant server — no runtime change needed.
Comment on lines
+241
to
+244
| internal NatsConnection GetTenantConnection(NatsTenant tenant) | ||
| { | ||
| return tenant.HasOwnConnection ? tenant.Connection ?? Connection : Connection; | ||
| } |
Comment on lines
+50
to
+51
| var msgId = _msgIdSource?.Invoke(envelope) ?? envelope.Id.ToString(); | ||
| return string.IsNullOrEmpty(msgId) ? null : new NatsJSPubOpts { MsgId = msgId }; |
Member
|
Heads up: this PR's CI is red on |
Member
|
Re-triggering CI: this PR's build was red on the |
This was referenced Jul 5, 2026
This was referenced Jul 14, 2026
Open
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Audit and hardening of the NATS transport for per-aggregate routing and server-side idempotency, plus a client upgrade. Closes the two spike blockers: static-only subjects and an inert JetStream duplicate window.
Changes
RoutingMode.ByTopic/Envelope.TopicName), mirroring RabbitMQ/Kafka/MQTT:PublishMessagesToNatsSubject<T>(Func<T,string>)for a subject computed per message.IMessageBus.BroadcastToTopicAsyncsupport (the ByTopic endpoint auto-enrolls).ISubjectResolverwired as an advanced envelope-level escape hatch.Nats-Msg-Idso the stream duplicate window actually deduplicates for external (non-Wolverine) consumers:DeduplicateUsing(Func<Envelope,string>)to project a domain identity.Nats-Msg-Idheader →DeduplicateUsing→Envelope.Id.DuplicateWindownow applied during stream provisioning (WithDeduplicationWindow/JetStreamDefaults.DuplicateWindow).NatsConnectionfollowing theLightweightCachepattern used by RabbitMQ/Azure.CompoundListener(one listener per tenant connection, tenant id stamped, completion routed back over the right connection) — mirrors the RabbitMQ / Azure Service Bus pattern.AckTerminateAsync(reason); warn when no dead-letter subject is configured (previously a silent ack + drop).CustomHeaders(AddOutgoingHeader),DefaultQueueGroup,NormalizeSubjects, theJetStreamDefaultstemplate, and JetStream domain / API-prefix through a centralized JetStream-context factory.ToJetStreamOpts(dead + buggy),TenantAwareNatsSender(superseded by frameworkTenantedSender),NatsEndpoint.BuildHeaders; collapse divergent stream provisioning ontoStreamConfiguration/JetStreamDefaults.Docs
docs/guide/messaging/transports/nats.mdupdated: dynamic subjects, dedup, dead-letter behavior, per-tenant connections + inbound, and an explicit note that NATS-native multi-tenancy is Accounts (a dedicated connection per account), while subject-prefix isolation is soft partitioning within one account.Testing
New integration tests in
Wolverine.Nats.Tests(135 total, all green on net10.0 against a JetStream broker; the per-tenant test spins up a second broker):PublishMessagesToNatsSubject<T>, two distinct computed subjects,BroadcastToTopicAsync, and theISubjectResolverescape hatch (exact concrete subject proven via a raw subscriber).Nats-Msg-Idheader honored, distinct messages not over-deduplicated.The full
wolverine.slnxRelease build introduces no new errors. (The pre-existingMicrosoft.OpenApiNU1903advisory is unrelated and addressed separately.)