JetStream leadership change fix (#180)

* JS publish reconnect and leader change fix * Code docs * Reverted unrelated changes
nats-io · Nov 6, 2023 · c3b988d · c3b988d
1 parent b576026
commit c3b988d
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs b/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs
@@ -284,6 +284,15 @@ protected override async ValueTask ReceiveInternalAsync(
                     else if (headers is { Code: 100, Message: NatsHeaders.Messages.IdleHeartbeat })
                     {
                     }
+                    else if (headers.Code == 409 && string.Equals(headers.MessageText, "Leadership Change", StringComparison.OrdinalIgnoreCase))
+                    {
+                        _logger.LogDebug(NatsJSLogEvents.LeadershipChange, "Leadership Change");
+                        lock (_pendingGate)
+                        {
+                            _pendingBytes = 0;
+                            _pendingMsgs = 0;
+                        }
+                    }
                     else if (headers.HasTerminalJSError())
                     {
                         _userMsgs.Writer.TryComplete(new NatsJSProtocolException($"JetStream server error: {headers.Code} {headers.MessageText}"));

diff --git a/src/NATS.Client.JetStream/NatsJSLogEvents.cs b/src/NATS.Client.JetStream/NatsJSLogEvents.cs
@@ -18,4 +18,5 @@ public static class NatsJSLogEvents
     public static readonly EventId NewDeliverySubject = new(2012, nameof(NewDeliverySubject));
     public static readonly EventId NewConsumerCreated = new(2013, nameof(NewConsumerCreated));
     public static readonly EventId Stopping = new(2014, nameof(Stopping));
+    public static readonly EventId LeadershipChange = new(2015, nameof(LeadershipChange));
 }