admin/ElsaWorkflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move NATS connection details from env vars onto the activity nodes

Browse Source 
Url/User/Password are now inputs on NatsPublish and NatsTrigger so authors
configure them in the workflow designer. The stored trigger stimulus carries
the connection info too, so messages from server A only fire workflows
designed to listen on server A. Dropped the AddNats singleton wiring and the
Nats section from appsettings.

A NatsConnectionCache singleton hands out one NatsConnection per unique
(url, user, password) so repeat publishes and each hosted-service consumer
loop reuse the same TCP session.
This commit is contained in:
EugeneTes
2026-09-02 09:57:04 +00:00
parent bd41609424
commit b2c2811933
9 changed files with 161 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
CLAUDE.md
View File

@@ -171,23 +171,32 @@ to its DSN.
## NATS activities
Three files under `src/Elsa.CustomServer/`:
Connection details (URL, user, password) live on the activity nodes themselves
— configured per-node in the workflow designer, not in appsettings or env vars.
Every stored trigger carries its own connection info; a `NatsConnectionCache`
singleton hands out one `NatsConnection` per unique `(url, user, password)` so
repeat invocations reuse the same TCP/TLS session.
- `Activities/NatsPublish.cs` — publisher `CodeActivity`. Inputs: `Subject`,
`Body`, optional `MessageId` (sets the `Nats-Msg-Id` header for JetStream
server-side dedup), optional `ExpectedStream` (fail-fast assertion). Outputs
`Sequence` and `Duplicate`.
- `Activities/NatsTrigger.cs` — trigger `Trigger<string>`. Inputs: `Stream`,
`Subject` (wildcards allowed). Outputs: `Result` = message body (UTF-8),
`ReceivedSubject`, `Headers`. Its trigger name / stimulus name is
`NATS.NatsTrigger`.
Files under `src/Elsa.CustomServer/`:
- `Activities/NatsPublish.cs` — publisher `CodeActivity`. Inputs: `Url`,
`User`, `Password`, `Subject`, `Body`, optional `MessageId` (sets the
`Nats-Msg-Id` header for JetStream server-side dedup), optional
`ExpectedStream` (fail-fast assertion). Outputs `Sequence` and `Duplicate`.
- `Activities/NatsTrigger.cs` — trigger `Trigger<string>`. Inputs: `Url`,
`User`, `Password`, `Stream`, `Subject` (wildcards allowed). Outputs:
`Result` = message body (UTF-8), `ReceivedSubject`, `Headers`. Its trigger
name / stimulus name is `NATS.NatsTrigger`; the stimulus record includes the
connection info so a message from server A only starts the workflows
designed to listen on server A.
- `Nats/NatsConnectionCache.cs` — singleton `(url, user, password) → NatsConnection` cache.
- `Nats/NatsTriggerHostedService.cs` — one `BackgroundService` that scans
`ITriggerStore` for stored triggers named `NATS.NatsTrigger`, spins up one
durable pull consumer per unique `(stream, subject)` pair, and dispatches
workflows via `IWorkflowDispatcher.DispatchAsync(DispatchTriggerWorkflowsRequest)`
as messages arrive. The mediator handler
`NatsTriggerIndexChangedHandler` re-runs the scan whenever
`WorkflowTriggersIndexed` fires (i.e. after publishing a workflow).
durable pull consumer per unique stimulus, and dispatches workflows via
`IWorkflowDispatcher.DispatchAsync(DispatchTriggerWorkflowsRequest)` as
messages arrive. The mediator handler `NatsTriggerIndexChangedHandler`
re-runs the scan whenever `WorkflowTriggersIndexed` fires (i.e. after
publishing a workflow).
Consumer model: shared durable pull consumer with `DurableName =
elsa-<sanitizedStream>-<sanitizedSubject>` and `AckPolicy.Explicit`. That means
@@ -195,14 +204,12 @@ if multiple Elsa instances are running, exactly one processes each message
(load-balanced), at-least-once. `MaxDeliver = 5`, `AckWait = 30s`,
`MaxAckPending = 256`. Failed dispatches NAK with a 5-second delay.
Configuration: `Nats:Url`, `Nats:User`, `Nats:Password` in appsettings
(env-var overrides `Nats__Url`, `Nats__User`, `Nats__Password`, or the
short `NATS_URL` / `NATS_USER` / `NATS_PASSWORD` fallbacks). Leave `Nats:Url`
empty to disable NATS entirely — the activities still register in the toolbox
but publishing throws and no consumers are started.
**Security note:** trigger stimuli — including the NATS password — are
persisted in the trigger store DB. Don't log the stimulus record verbatim, and
treat the trigger store with the same care as any secret storage.
Packages: `NATS.Net` + `NATS.Client.Hosting` (v2 client). The v1 `NATS.Client`
package does not support `wss://` from .NET — do not use it.
Package: `NATS.Net` (v2 client). The v1 `NATS.Client` package does not support
`wss://` from .NET — do not use it.
## Session-specific reminders

19
src/Elsa.CustomServer/Activities/NatsPublish.cs
View File

@@ -1,4 +1,5 @@
using System.Text;
using Elsa.CustomServer.Nats;
using Elsa.Extensions;
using Elsa.Workflows;
using Elsa.Workflows.Attributes;
@@ -12,6 +13,15 @@ namespace Elsa.CustomServer.Activities;
[Activity("NATS", "NATS", "Publish a message to a NATS JetStream subject.", DisplayName = "NATS Publish", Kind = ActivityKind.Task)]
public class NatsPublish : CodeActivity
{
[Input(Description = "NATS server URL (nats://, tls://, ws://, wss://).", Category = "Connection")]
public Input<string> Url { get; set; } = default!;
[Input(Description = "Username (optional).", Category = "Connection")]
public Input<string?> User { get; set; } = new(default(string?));
[Input(Description = "Password (optional).", Category = "Connection")]
public Input<string?> Password { get; set; } = new(default(string?));
[Input(Description = "Subject to publish to (must be bound to a JetStream stream).")]
public Input<string> Subject { get; set; } = default!;
@@ -32,6 +42,10 @@ public class NatsPublish : CodeActivity
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var url = context.Get(Url);
if (string.IsNullOrWhiteSpace(url))
throw new InvalidOperationException("NatsPublish: Url is required.");
var subject = context.Get(Subject);
if (string.IsNullOrWhiteSpace(subject))
throw new InvalidOperationException("NatsPublish: Subject is required.");
@@ -40,8 +54,9 @@ public class NatsPublish : CodeActivity
var messageId = context.Get(MessageId);
var expectedStream = context.Get(ExpectedStream);
var nats = context.GetRequiredService<INatsConnection>();
var js = new NatsJSContext((NatsConnection)nats);
var pool = context.GetRequiredService<NatsConnectionCache>();
var nats = pool.Get(url, context.Get(User), context.Get(Password));
var js = new NatsJSContext(nats);
NatsHeaders? headers = null;
if (!string.IsNullOrEmpty(messageId))

28
src/Elsa.CustomServer/Activities/NatsTrigger.cs
View File

@@ -1,4 +1,5 @@
using Elsa.CustomServer.Nats;
using Elsa.Expressions.Models;
using Elsa.Extensions;
using Elsa.Workflows;
using Elsa.Workflows.Attributes;
@@ -18,10 +19,19 @@ public class NatsTrigger : Trigger<string>
public const string SubjectInputKey = "NatsSubject";
public const string HeadersInputKey = "NatsHeaders";
[Input(Description = "NATS server URL (nats://, tls://, ws://, wss://).", Category = "Connection")]
public Input<string> Url { get; set; } = default!;
[Input(Description = "Username (optional).", Category = "Connection")]
public Input<string?> User { get; set; } = new(default(string?));
[Input(Description = "Password (optional).", Category = "Connection")]
public Input<string?> Password { get; set; } = new(default(string?));
[Input(Description = "JetStream name that owns the subject.")]
public Input<string> Stream { get; set; } = default!;
[Input(Description = "Subject to subscribe to. Wildcards are allowed: 'orders.*' matches one token, 'orders.>' matches one or more.")]
[Input(Description = "Subject to subscribe to. Wildcards allowed: 'orders.*' matches one token, 'orders.>' matches one or more.")]
public Input<string> Subject { get; set; } = default!;
[Output(Description = "The concrete subject the message was published to (useful with wildcards).")]
@@ -33,9 +43,7 @@ public class NatsTrigger : Trigger<string>
protected override IEnumerable<object> GetTriggerPayloads(TriggerIndexingContext context)
{
context.TriggerName = StimulusName;
var stream = Stream.Get(context.ExpressionExecutionContext);
var subject = Subject.Get(context.ExpressionExecutionContext);
return [new NatsTriggerStimulus(stream, subject)];
return [BuildStimulus(context.ExpressionExecutionContext)];
}
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
@@ -48,11 +56,17 @@ public class NatsTrigger : Trigger<string>
}
// Mid-workflow encounter (rare) — park on a bookmark so a matching message can resume.
var stream = Stream.Get(context);
var subject = Subject.Get(context);
context.CreateBookmark(new NatsTriggerStimulus(stream, subject), OnResumeAsync, includeActivityInstanceId: false);
context.CreateBookmark(BuildStimulus(context.ExpressionExecutionContext), OnResumeAsync, includeActivityInstanceId: false);
}
private NatsTriggerStimulus BuildStimulus(ExpressionExecutionContext context) =>
new(
Url.Get(context) ?? string.Empty,
User.Get(context) ?? string.Empty,
Password.Get(context) ?? string.Empty,
Stream.Get(context) ?? string.Empty,
Subject.Get(context) ?? string.Empty);
private async ValueTask OnResumeAsync(ActivityExecutionContext context)
{
EmitOutputs(context);

4
src/Elsa.CustomServer/Elsa.CustomServer.csproj
View File

@@ -20,9 +20,9 @@
<PackageReference Include="Elsa.CSharp" Version="3.5.3" />
<PackageReference Include="Elsa.JavaScript" Version="3.5.3" />
<PackageReference Include="Elsa.Liquid" Version="3.5.3" />
<!-- NATS.Net v2 client + hosting extensions for JetStream publish/subscribe. -->
<!-- NATS.Net v2 client for JetStream publish/subscribe. Connection details
are configured per-activity in the workflow designer, not via DI. -->
<PackageReference Include="NATS.Net" Version="2.5.7" />
<PackageReference Include="NATS.Client.Hosting" Version="2.5.7" />
<!-- Hosts the Blazor WebAssembly Studio in the same process. -->
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="9.0.13" />
</ItemGroup>

46
src/Elsa.CustomServer/Nats/NatsConnectionCache.cs Normal file
View File

@@ -0,0 +1,46 @@
using System.Collections.Concurrent;
using NATS.Client.Core;
namespace Elsa.CustomServer.Nats;
// Lazily creates one NatsConnection per (url, user, password) tuple and reuses
// it for the process lifetime. Registered as a singleton so both the publish
// activity and the trigger subscriber share the same pool.
public sealed class NatsConnectionCache : IAsyncDisposable
{
private readonly ConcurrentDictionary<string, Lazy<NatsConnection>> _connections = new();
public NatsConnection Get(string url, string? user, string? password)
{
if (string.IsNullOrWhiteSpace(url))
throw new ArgumentException("NATS Url is required.", nameof(url));
var key = $"{url}|{user}|{password}";
var lazy = _connections.GetOrAdd(key, _ => new Lazy<NatsConnection>(() =>
{
var opts = NatsOpts.Default with
{
Url = url,
Name = "elsa-workflows",
AuthOpts = new NatsAuthOpts
{
Username = string.IsNullOrEmpty(user) ? null : user,
Password = string.IsNullOrEmpty(password) ? null : password,
},
};
return new NatsConnection(opts);
}));
return lazy.Value;
}
public async ValueTask DisposeAsync()
{
foreach (var entry in _connections.Values)
{
if (!entry.IsValueCreated) continue;
try { await entry.Value.DisposeAsync(); } catch { /* best-effort */ }
}
_connections.Clear();
}
}

53
src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs
View File

@@ -4,25 +4,22 @@ using System.Text.Json;
using Elsa.CustomServer.Activities;
using Elsa.Workflows.Runtime;
using Elsa.Workflows.Runtime.Filters;
using Elsa.Workflows.Runtime.Messages;
using Elsa.Workflows.Runtime.Requests;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
namespace Elsa.CustomServer.Nats;
// Consumes NATS JetStream on behalf of NatsTrigger-typed workflow triggers.
// On startup (and again whenever the trigger index changes) it groups the stored
// triggers by (stream, subject), spins up one durable pull consumer per unique
// pair, and dispatches a workflow per received message. Uses a stable, hashed
// DurableName that includes the stream + subject so multiple Elsa instances form
// a load-balanced consumer group for each subject.
// Each stored trigger carries its own (Url, User, Password, Stream, Subject)
// stimulus; the service spins up one durable pull consumer per unique stimulus
// and dispatches a workflow per received message. Consumers use a deterministic
// DurableName so multiple Elsa instances form a load-balanced group.
public sealed class NatsTriggerHostedService(
INatsConnection nats,
NatsConnectionCache connectionCache,
IServiceProvider services,
ILogger<NatsTriggerHostedService> log)
: BackgroundService
@@ -53,7 +50,7 @@ public sealed class NatsTriggerHostedService(
foreach (var t in stored)
{
var stim = TryDeserializeStimulus(t.Payload);
if (stim is null)
if (stim is null || string.IsNullOrWhiteSpace(stim.Url) || string.IsNullOrWhiteSpace(stim.Stream) || string.IsNullOrWhiteSpace(stim.Subject))
continue;
desired.Add(stim);
}
@@ -64,7 +61,7 @@ public sealed class NatsTriggerHostedService(
if (desired.Contains(key)) continue;
if (_subscriptions.TryRemove(key, out var cts))
{
log.LogInformation("Stopping NATS consumer for stream={Stream} subject={Subject}", key.Stream, key.Subject);
log.LogInformation("Stopping NATS consumer for url={Url} stream={Stream} subject={Subject}", key.Url, key.Stream, key.Subject);
cts.Cancel();
cts.Dispose();
}
@@ -77,8 +74,8 @@ public sealed class NatsTriggerHostedService(
var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
if (_subscriptions.TryAdd(key, cts))
{
log.LogInformation("Starting NATS consumer for stream={Stream} subject={Subject}", key.Stream, key.Subject);
_ = Task.Run(() => ConsumeLoopAsync(key.Stream, key.Subject, cts.Token), CancellationToken.None);
log.LogInformation("Starting NATS consumer for url={Url} stream={Stream} subject={Subject}", key.Url, key.Stream, key.Subject);
_ = Task.Run(() => ConsumeLoopAsync(key, cts.Token), CancellationToken.None);
}
}
}
@@ -103,27 +100,27 @@ public sealed class NatsTriggerHostedService(
}
}
private async Task ConsumeLoopAsync(string stream, string subject, CancellationToken ct)
private async Task ConsumeLoopAsync(NatsTriggerStimulus stim, CancellationToken ct)
{
// Deterministic consumer name so every Elsa instance joins the same group.
// Sanitize because NATS consumer names disallow '.', '*', '>', whitespace.
var consumerName = $"elsa-{Sanitize(stream)}-{Sanitize(subject)}";
var consumerName = $"elsa-{Sanitize(stim.Stream)}-{Sanitize(stim.Subject)}";
var js = new NatsJSContext((NatsConnection)nats);
// Reconnect-forever outer loop — if the consumer creation or consume stream
// faults, back off and try again. NatsConnection handles TCP-level reconnects
// itself; this catches higher-level errors (server restart, permissions).
// Reconnect-forever outer loop — catches higher-level errors (server restart, permissions).
// NatsConnection handles TCP-level reconnects itself.
while (!ct.IsCancellationRequested)
{
try
{
var consumer = await js.CreateOrUpdateConsumerAsync(stream, new ConsumerConfig
var nats = connectionCache.Get(stim.Url, stim.User, stim.Password);
var js = new NatsJSContext(nats);
var consumer = await js.CreateOrUpdateConsumerAsync(stim.Stream, new ConsumerConfig
{
DurableName = consumerName,
AckPolicy = ConsumerConfigAckPolicy.Explicit,
DeliverPolicy = ConsumerConfigDeliverPolicy.New, // only applied on first-time creation
FilterSubject = subject,
FilterSubject = stim.Subject,
MaxAckPending = 256,
AckWait = TimeSpan.FromSeconds(30),
MaxDeliver = 5,
@@ -133,7 +130,7 @@ public sealed class NatsTriggerHostedService(
{
try
{
await DispatchAsync(stream, subject, msg, ct);
await DispatchAsync(stim, msg, ct);
await msg.AckAsync(cancellationToken: ct);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
@@ -142,14 +139,14 @@ public sealed class NatsTriggerHostedService(
}
catch (Exception ex)
{
log.LogError(ex, "NATS dispatch failed for stream={Stream} subject={Subject} seq={Seq}; NAKing", stream, msg.Subject, msg.Metadata?.Sequence.Stream);
log.LogError(ex, "NATS dispatch failed for stream={Stream} subject={Subject} seq={Seq}; NAKing", stim.Stream, msg.Subject, msg.Metadata?.Sequence.Stream);
try
{
await msg.NakAsync(delay: TimeSpan.FromSeconds(5), cancellationToken: ct);
}
catch (Exception nakEx)
{
log.LogWarning(nakEx, "NAK failed for stream={Stream} subject={Subject}", stream, msg.Subject);
log.LogWarning(nakEx, "NAK failed for stream={Stream} subject={Subject}", stim.Stream, msg.Subject);
}
}
}
@@ -160,14 +157,14 @@ public sealed class NatsTriggerHostedService(
}
catch (Exception ex)
{
log.LogError(ex, "NATS consume loop crashed for stream={Stream} subject={Subject}; retrying in 5s", stream, subject);
log.LogError(ex, "NATS consume loop crashed for url={Url} stream={Stream} subject={Subject}; retrying in 5s", stim.Url, stim.Stream, stim.Subject);
try { await Task.Delay(TimeSpan.FromSeconds(5), ct); }
catch (OperationCanceledException) { return; }
}
}
}
private async Task DispatchAsync(string stream, string subject, NatsJSMsg<byte[]> msg, CancellationToken ct)
private async Task DispatchAsync(NatsTriggerStimulus stim, NatsJSMsg<byte[]> msg, CancellationToken ct)
{
using var scope = services.CreateScope();
var dispatcher = scope.ServiceProvider.GetRequiredService<IWorkflowDispatcher>();
@@ -180,7 +177,7 @@ public sealed class NatsTriggerHostedService(
headers[key] = h[key].ToString();
}
var request = new DispatchTriggerWorkflowsRequest(NatsTrigger.StimulusName, new NatsTriggerStimulus(stream, subject))
var request = new DispatchTriggerWorkflowsRequest(NatsTrigger.StimulusName, stim)
{
Input = new Dictionary<string, object>
{
@@ -192,7 +189,7 @@ public sealed class NatsTriggerHostedService(
await dispatcher.DispatchAsync(request, options: null!, cancellationToken: ct);
log.LogInformation("NATS → dispatched trigger stream={Stream} subject={Subject} seq={Seq}",
stream, msg.Subject, msg.Metadata?.Sequence.Stream);
stim.Stream, msg.Subject, msg.Metadata?.Sequence.Stream);
}
private static string Sanitize(string value) =>

15
src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs
View File

@@ -2,5 +2,16 @@ namespace Elsa.CustomServer.Nats;
// Stimulus used both as the indexed trigger payload and as the bookmark payload
// when a NATS message dispatches a workflow. Structural equality (record) is what
// lets the runtime match published triggers against incoming messages.
public sealed record NatsTriggerStimulus(string Stream, string Subject);
// lets the runtime match published triggers against incoming messages, so the
// connection details are part of the key: a message pulled from server A doesn't
// start a workflow that was designed to listen on server B.
//
// Note: credentials are persisted in the trigger store as part of this payload,
// which is the trade-off for letting workflow authors configure NATS in the
// designer. Don't log this record verbatim.
public sealed record NatsTriggerStimulus(
string Url,
string User,
string Password,
string Stream,
string Subject);

38
src/Elsa.CustomServer/Program.cs
View File

@@ -7,8 +7,6 @@ using Elsa.Extensions;
using Elsa.Mediator.Extensions;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.HttpOverrides;
using NATS.Client.Core;
using NATS.Client.Hosting;
var builder = WebApplication.CreateBuilder(args);
builder.WebHost.UseStaticWebAssets();
@@ -42,33 +40,15 @@ builder.Services.AddElsa(elsa =>
.UseLiquid(liquid => liquid.FluidOptions = options => options.Encoder = HtmlEncoder.Default);
});
// NATS: single shared NatsConnection wired from Nats:Url / Nats:User / Nats:Password
// (with env-var fallback NATS_URL / NATS_USER / NATS_PASSWORD). The AddNats hosting
// extension registers INatsConnection as a singleton; wss://, tls://, nats:// are
// all handled transparently based on the URL scheme.
var natsSection = configuration.GetSection("Nats");
var natsUrl = natsSection["Url"] ?? Environment.GetEnvironmentVariable("NATS_URL");
var natsUser = natsSection["User"] ?? Environment.GetEnvironmentVariable("NATS_USER");
var natsPassword = natsSection["Password"] ?? Environment.GetEnvironmentVariable("NATS_PASSWORD");
if (!string.IsNullOrWhiteSpace(natsUrl))
{
builder.Services.AddNats(configureOpts: opts => opts with
{
Url = natsUrl,
Name = "elsa-workflows",
AuthOpts = new NatsAuthOpts
{
Username = string.IsNullOrEmpty(natsUser) ? null : natsUser,
Password = string.IsNullOrEmpty(natsPassword) ? null : natsPassword,
},
});
// Hosted subscriber that fans workflow-trigger dispatches out from stored NatsTrigger definitions.
builder.Services.AddSingleton<NatsTriggerHostedService>();
builder.Services.AddHostedService(sp => sp.GetRequiredService<NatsTriggerHostedService>());
builder.Services.AddNotificationHandler<NatsTriggerIndexChangedHandler, Elsa.Workflows.Runtime.Notifications.WorkflowTriggersIndexed>();
}
// NATS: connection details (URL/user/password) live on the NatsPublish / NatsTrigger
// activity nodes themselves — configured in the workflow designer. A single pool
// caches one NatsConnection per (url, user, password) so repeat invocations reuse
// the same connection. The hosted service reads stored NatsTrigger stimuli and
// spins up one durable pull consumer per unique connection+subject.
builder.Services.AddSingleton<NatsConnectionCache>();
builder.Services.AddSingleton<NatsTriggerHostedService>();
builder.Services.AddHostedService(sp => sp.GetRequiredService<NatsTriggerHostedService>());
builder.Services.AddNotificationHandler<NatsTriggerIndexChangedHandler, Elsa.Workflows.Runtime.Notifications.WorkflowTriggersIndexed>();
builder.Services.AddCors(cors => cors.AddDefaultPolicy(p => p
.AllowAnyHeader()

5
src/Elsa.CustomServer/appsettings.json
View File

@@ -16,11 +16,6 @@
"Http": {
"BasePath": "/workflows"
},
"Nats": {
"Url": "",
"User": "",
"Password": ""
},
"Identity": {
"Tokens": {
"SigningKey": "CHANGE_ME_VIA_ENV_Identity__Tokens__SigningKey",