Nats consumer retry and replicas

[NSTA2]

NATS streaming consumer permanent failure after transient JetStream errors

Problem

When running Orleans against a multi-node NATS JetStream cluster, a rolling restart of NATS nodes (routine maintenance, crash recovery, scaling) causes permanent consumer failure with no recovery path short of restarting the Orleans silo.

Two issues combine to produce this:

1. No retry on consumer initialization

NatsStreamConsumer.Initialize() is called exactly once during startup. If it fails due to a transient error — timeout during JetStream leader election, network blip, temporary unavailability — the internal _consumer field stays null permanently. Every subsequent GetMessages() poll (~100ms) logs at Error level:

Internal NATS Consumer is not initialized. Provider: {Provider} | Stream: {Stream} | Partition: {Partition}.

…and returns empty, indefinitely, with no self-healing path.

2. Hardcoded R1 streams

NatsConnectionManager.Initialize() creates the JetStream stream without setting NumReplicas, defaulting to R1 (single replica). R1 streams have exactly one leader; any node restart makes the stream temporarily unavailable during leader election — which is the trigger for bug #1.

Combined effect: After a NATS rolling update, Orleans consumers enter a permanent error loop on every poll cycle, producing a flood of error logs and zero message delivery until the entire Orleans pod is restarted.

Root Cause

NATS node restart
  → R1 stream leader temporarily unavailable
    → NatsStreamConsumer.Initialize() throws (timeout / leader election)
      → _consumer stays null permanently
        → Every GetMessages() poll logs Error + returns empty
          → No recovery without full silo restart

Changes

File	Change
NatsStreamConsumer.cs	When _consumer is null in GetMessages(), attempt lazy re-initialization before returning empty. Piggybacks on the existing Orleans pulling agent poll cadence. Log level changed from Error to Warning — transient retries during rolling updates are expected, not permanent failures.
NatsOptions.cs	Added NumReplicas property (default 1, backward compatible). Added validation in NatsStreamOptionsValidator ensuring NumReplicas >= 1.
NatsConnectionManager.cs	Passes NumReplicas to StreamConfig when creating JetStream streams. Handles NATS error code 10058 (stream exists with different config) by attempting an in-place UpdateStreamAsync, enabling replica count upgrades without manual stream deletion.
NatsOptionsTests.cs (new)	Unit tests for validator (invalid/valid NumReplicas, missing/empty StreamName), default value assertion, and an integration test verifying NumReplicas flows through to JetStream stream config.

Usage

siloBuilder.AddNatsStreams("my-provider", options =>
{
    options.StreamName = "my-stream";
    options.NumReplicas = 3; // R3 for production HA clusters (≥ 3 NATS nodes)
});

Backward Compatibility

• NumReplicas defaults to 1 — existing deployments are unaffected.
• The retry in GetMessages() is purely additive — previously broken consumers now self-heal.
• Error 10058 handling is additive — previously this was an unhandled exception that crashed the adapter factory.

Testing

• All 50 existing NATS integration tests pass (stream, subscription multiplicity, client stream).
• 10 new tests added: validator unit tests (no NATS required) + integration test verifying NumReplicas is applied to JetStream stream config.
• The 2 pre-existing NatsAdapterTests.SendAndReceiveFromNats failures are unrelated (cache cursor requests SeqNum=0 but JetStream sequences start at 1).
• R3 integration testing requires a multi-node NATS cluster and should be validated in CI with a 3-node docker-compose setup.

Microsoft Reviewers: Open in CodeFlow

[NSTA2]

@dotnet-policy-service agree company="Microsoft"

[Copilot]

Pull request overview

This PR improves resiliency of the Orleans NATS JetStream streaming provider in multi-node NATS clusters by (1) enabling consumer self-healing after transient JetStream errors and (2) allowing stream replica count configuration so rolling restarts don’t permanently break consumption.

Changes:

• Add lazy re-initialization of NatsStreamConsumer when _consumer is null during polling.
• Introduce NatsOptions.NumReplicas with validation and apply it when creating JetStream streams (plus attempt stream update on config mismatch).
• Add tests covering NumReplicas defaults/validation and a JetStream config assertion.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

File	Description
test/Extensions/Orleans.Streaming.NATS.Tests/NatsOptionsTests.cs	Adds unit tests for options validation and a JetStream-facing test related to replicas.
src/Orleans.Streaming.NATS/Providers/NatsStreamConsumer.cs	Adds retry-on-poll initialization logic and lowers severity of “not initialized” logging.
src/Orleans.Streaming.NATS/Providers/NatsConnectionManager.cs	Plumbs NumReplicas into stream creation and attempts UpdateStreamAsync on config mismatch.
src/Orleans.Streaming.NATS/NatsOptions.cs	Adds NumReplicas option and validates it is at least 1.

[Copilot]

GetMessages logs a warning on every poll cycle while the consumer is uninitialized. During a longer outage/misconfiguration this can still produce very high log volume (poll cadence is ~100ms). Consider adding a backoff/rate-limit (eg, log once then periodically, or only log after N consecutive failures), while still retrying initialization each cycle.

[Copilot]

The XML docs state NumReplicas "must be an odd number" and "cannot exceed the number of NATS nodes", but the validator only enforces >= 1. Either relax the docs to match what is actually enforced, or add validation for the odd-number requirement (and consider how to handle/enforce the upper bound, if possible).

[Copilot]

streamConfig is constructed twice (CreateStreamAsync + UpdateStreamAsync) with the same fields. Consider extracting a private helper/local function to build the config once to avoid future drift (eg, if Retention/SubjectTransform changes later but only one path is updated).

[Copilot]

NumReplicas_IsAppliedToJetStreamConfig doesn't exercise the provider path which was changed (NatsOptions -> NatsConnectionManager -> StreamConfig). The test currently hard-codes NumReplicas = 1 directly on StreamConfig, so it will pass even if _options.NumReplicas is never applied by the provider. Consider updating this to initialize a NatsConnectionManager with options.NumReplicas and asserting the resulting stream info, or rename the test so it reflects what is actually being validated.