diff --git a/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs b/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs index c33cb6c09..088983a0e 100644 --- a/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs +++ b/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs @@ -116,6 +116,28 @@ private static string ParseError(in ReadOnlySequence errorSlice) return Encoding.UTF8.GetString(errorSlice.Slice(6, errorSlice.Length - 7)); } + private void HandleServerError(string error) + { + var args = new NatsServerErrorEventArgs(error); + + // Server-initiated graceful disconnects (expiry/revoke/stale) are + // expected and recovered by reconnect; log at Debug so they don't + // show up as errors. Callers that need visibility can subscribe to + // the ServerError event. Matches nats.go which doesn't log at all + // for these and surfaces them via AsyncErrorCB. + var level = args.Kind switch + { + NatsServerErrorKind.AuthenticationExpired => LogLevel.Debug, + NatsServerErrorKind.AccountAuthenticationExpired => LogLevel.Debug, + NatsServerErrorKind.AuthenticationRevoked => LogLevel.Debug, + NatsServerErrorKind.StaleConnection => LogLevel.Debug, + _ => LogLevel.Error, + }; + _logger.Log(level, NatsLogEvents.Protocol, "Server error {Error}", error); + _connection.PushEvent(NatsEvent.ServerError, args); + _waitForPongOrErrorSignal.TrySetObservedException(new NatsServerException(error)); + } + private async Task ReadLoopAsync() { while (true) @@ -402,18 +424,16 @@ private async ValueTask> DispatchCommandAsync(int code, R _socketReader.AdvanceTo(buffer.Start); var newBuffer = await _socketReader.ReadUntilReceiveNewLineAsync().ConfigureAwait(false); var newPosition = newBuffer.PositionOf((byte)'\n'); - var error = ParseError(newBuffer.Slice(0, newBuffer.GetOffset(newPosition!.Value) - 1)); - _logger.LogError(NatsLogEvents.Protocol, "Server error {Error}", error); - _connection.PushEvent(NatsEvent.ServerError, new NatsServerErrorEventArgs(error)); - _waitForPongOrErrorSignal.TrySetObservedException(new NatsServerException(error)); + var lineWithCR = newBuffer.Slice(0, newPosition!.Value); + var error = ParseError(lineWithCR.Slice(0, lineWithCR.Length - 1)); + HandleServerError(error); return newBuffer.Slice(newBuffer.GetPosition(1, newPosition!.Value)); } else { - var error = ParseError(buffer.Slice(0, buffer.GetOffset(position.Value) - 1)); - _logger.LogError(NatsLogEvents.Protocol, "Server error {Error}", error); - _connection.PushEvent(NatsEvent.ServerError, new NatsServerErrorEventArgs(error)); - _waitForPongOrErrorSignal.TrySetObservedException(new NatsServerException(error)); + var lineWithCR = buffer.Slice(0, position.Value); + var error = ParseError(lineWithCR.Slice(0, lineWithCR.Length - 1)); + HandleServerError(error); return buffer.Slice(buffer.GetPosition(1, position.Value)); } } diff --git a/tests/NATS.Client.Core.Tests/AuthErrorTest.cs b/tests/NATS.Client.Core.Tests/AuthErrorTest.cs index 8b278901c..165bc5397 100644 --- a/tests/NATS.Client.Core.Tests/AuthErrorTest.cs +++ b/tests/NATS.Client.Core.Tests/AuthErrorTest.cs @@ -10,6 +10,96 @@ public class AuthErrorTest public AuthErrorTest(ITestOutputHelper output) => _output = output; + // Regression for an established connection receiving -ERR mid-session + // (e.g. JWT expiry, auth callout token rotation). The client should + // reconnect without producing UnobservedTaskException from the + // server exception constructed inside the read loop, and recoverable + // server-initiated graceful disconnects should log at Debug rather + // than Error. + [Theory] + [InlineData("User Authentication Expired", LogLevel.Debug)] + [InlineData("Account Authentication Expired", LogLevel.Debug)] + [InlineData("User Authentication Revoked", LogLevel.Debug)] + [InlineData("Stale Connection", LogLevel.Debug)] + [InlineData("Permissions Violation for Publish to foo", LogLevel.Error)] + public async Task Mid_session_server_error_reconnects_cleanly(string serverError, LogLevel expectedLevel) + { + var unobservedThrown = 0; + Exception? unobservedException = null; + EventHandler handler = (_, e) => + { + unobservedException = e.Exception; + Interlocked.Increment(ref unobservedThrown); + }; + + var logs = new InMemoryTestLoggerFactory(LogLevel.Trace, m => _output.WriteLine($"[LOG {m.LogLevel}] {m.Message}")); + + TaskScheduler.UnobservedTaskException += handler; + try + { + var pingCount = 0; + var reconnected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var server = new MockServer( + handler: async (client, cmd) => + { + if (cmd.Name != "PING") + return; + + var n = Interlocked.Increment(ref pingCount); + if (n == 1) + { + // First PING after CONNECT: auto-PONG already sent. + // Now push a mid-session -ERR and close the socket so + // the client must reconnect. + await client.Writer.WriteAsync($"-ERR '{serverError}'\r\n"); + await client.Writer.FlushAsync(); + client.Close(); + } + else if (n == 2) + { + // Second connection's initial PING: client is back up. + reconnected.TrySetResult(); + } + }, + logger: m => _output.WriteLine(m)); + + await server.Ready; + + await using (var nats = new NatsConnection(new NatsOpts + { + Url = server.Url, + LoggerFactory = logs, + ReconnectWaitMin = TimeSpan.FromMilliseconds(50), + ReconnectWaitMax = TimeSpan.FromMilliseconds(100), + })) + { + await nats.ConnectAsync(); + await reconnected.Task.WaitAsync(TimeSpan.FromSeconds(10)); + } + + // Give any orphaned exceptions a chance to surface. + await Task.Delay(100); + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + } + finally + { + TaskScheduler.UnobservedTaskException -= handler; + } + + Assert.True( + unobservedThrown == 0, + $"Expected no unobserved exceptions, got {unobservedThrown}: {unobservedException}"); + + var serverErrorLogs = logs.Logs + .Where(m => m.Message.StartsWith("Server error ")) + .ToList(); + Assert.NotEmpty(serverErrorLogs); + Assert.All(serverErrorLogs, m => Assert.Equal(expectedLevel, m.LogLevel)); + } + // [SkipOnPlatform("WINDOWS", "doesn't support HUP signal")] [Fact] public async Task Auth_err_twice_will_stop_retries()