Added verification of the consumer sequence number for pull ordered consumers (#981)

* Added verification of the consumer sequence number for pull ordered consumer

* Fix an ordered consumer retry logic for MaxBytes mode

* Add tests minor fixes

* Fix msg count, size check

* Fix test

---------

Co-authored-by: Ziya Suzen <ziya@suzen.net>

Author: sergeimam

src/NATS.Client.JetStream/NatsJSOrderedConsumer.cs

@@ -79,6 +79,7 @@ public async IAsyncEnumerable<NatsJSMsg<T>> ConsumeAsync<T>(
                 consumerName = consumer.Info.Name;
                 _logger.LogInformation(NatsJSLogEvents.NewConsumer, "Created {ConsumerName} with sequence {Seq}", consumerName, seq);
 
+                ulong cseq = 0;
                 NatsJSProtocolException? protocolException = default;
 
                 await using (var cc = await consumer.OrderedConsumeInternalAsync(serializer, opts, cancellationToken))
@@ -126,7 +127,15 @@ public async IAsyncEnumerable<NatsJSMsg<T>> ConsumeAsync<T>(
                             if (msg.Metadata is not { } metadata)
                                 continue;
 
+                            var expected = cseq + 1;
+                            if (metadata.Sequence.Consumer != expected)
+                            {
+                                _logger.LogWarning(NatsJSLogEvents.Retry, "Consumer sequence mismatch. Expected {Expected}, was {SequenceConsumer}. Retrying...", expected, metadata.Sequence.Consumer);
+                                goto CONSUME_LOOP;
+                            }
+
                             seq = metadata.Sequence.Stream;
+                            cseq = metadata.Sequence.Consumer;
 
                             yield return msg;
                         }
@@ -180,23 +189,65 @@ public async IAsyncEnumerable<NatsJSMsg<T>> FetchAsync<T>(
         [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_cancellationToken, cancellationToken).Token;
+        var processed = 0;
+        var bytesProcessed = 0;
 
-        var consumer = await RecreateConsumer(_fetchConsumerName, _fetchSeq, cancellationToken);
-        _fetchConsumerName = consumer.Info.Name;
-
-        await foreach (var msg in consumer.FetchAsync(opts, serializer, cancellationToken))
+        var retry = 0;
+        while (!cancellationToken.IsCancellationRequested)
         {
-            if (msg.Metadata is not { } metadata)
-                continue;
+            if ((opts.MaxMsgs.HasValue && processed >= opts.MaxMsgs) || (opts.MaxBytes.HasValue && bytesProcessed >= opts.MaxBytes))
+                yield break;
 
-            _fetchSeq = metadata.Sequence.Stream;
-            yield return msg;
-        }
+            var mismatch = false;
+            ulong cseq = 0;
+
+            var consumer = await RecreateConsumer(_fetchConsumerName, _fetchSeq, cancellationToken);
+            _fetchConsumerName = consumer.Info.Name;
+
+            try
+            {
+                var fetchOpts = opts with { MaxMsgs = opts.MaxMsgs - processed, MaxBytes = opts.MaxBytes - bytesProcessed };
+
+                await foreach (var msg in consumer.FetchAsync(fetchOpts, serializer, cancellationToken))
+                {
+                    if (msg.Metadata is not { } metadata)
+                        continue;
+
+                    var expected = cseq + 1;
+                    if (metadata.Sequence.Consumer != expected)
+                    {
+                        _logger.LogWarning(NatsJSLogEvents.Retry, "Consumer sequence mismatch. Expected {Expected}, was {SequenceConsumer}. Retrying...", expected, metadata.Sequence.Consumer);
+                        mismatch = true;
+                        break;
+                    }
+
+                    _fetchSeq = metadata.Sequence.Stream;
+                    cseq = metadata.Sequence.Consumer;
+
+                    processed++;
+                    bytesProcessed += msg.Size;
+
+                    yield return msg;
+                }
+            }
+            finally
+            {
+                var deleted = await TryDeleteConsumer(_fetchConsumerName, cancellationToken);
+
+                if (deleted)
+                    _fetchConsumerName = string.Empty;
+            }
 
-        var deleted = await TryDeleteConsumer(_fetchConsumerName, cancellationToken);
+            if (!mismatch)
+                yield break;
+
+            if (retry == _opts.MaxResetAttempts)
+            {
+                throw new NatsJSException("Maximum number of retry attempts reached.");
+            }
 
-        if (deleted)
-            _fetchConsumerName = string.Empty;
+            await _context.Connection.Opts.BackoffWithJitterAsync(retry++, cancellationToken);
+        }
     }
 
     /// <inheritdoc />
@@ -206,26 +257,63 @@ public async IAsyncEnumerable<NatsJSMsg<T>> FetchNoWaitAsync<T>(
         [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         cancellationToken = CancellationTokenSource.CreateLinkedTokenSource(_cancellationToken, cancellationToken).Token;
+        var processed = 0;
+        var bytesProcessed = 0;
 
-        var consumer = await RecreateConsumer(_fetchConsumerName, _fetchSeq, cancellationToken);
-        _fetchConsumerName = consumer.Info.Name;
-        try
+        var retry = 0;
+        while (!cancellationToken.IsCancellationRequested)
         {
-            await foreach (var msg in consumer.FetchNoWaitAsync(opts, serializer, cancellationToken))
+            if ((opts.MaxMsgs.HasValue && processed >= opts.MaxMsgs) || (opts.MaxBytes.HasValue && bytesProcessed >= opts.MaxBytes))
+                yield break;
+
+            var mismatch = false;
+            ulong cseq = 0;
+            var consumer = await RecreateConsumer(_fetchConsumerName, _fetchSeq, cancellationToken);
+            _fetchConsumerName = consumer.Info.Name;
+
+            try
+            {
+                var fetchOpts = opts with { MaxMsgs = opts.MaxMsgs - processed, MaxBytes = opts.MaxBytes - bytesProcessed };
+
+                await foreach (var msg in consumer.FetchNoWaitAsync(fetchOpts, serializer, cancellationToken))
+                {
+                    if (msg.Metadata is not { } metadata)
+                        continue;
+
+                    var expected = cseq + 1;
+                    if (metadata.Sequence.Consumer != expected)
+                    {
+                        _logger.LogWarning(NatsJSLogEvents.Retry, "Consumer sequence mismatch. Expected {Expected}, was {SequenceConsumer}. Retrying...", expected, metadata.Sequence.Consumer);
+                        mismatch = true;
+                        break;
+                    }
+
+                    _fetchSeq = metadata.Sequence.Stream;
+                    cseq = metadata.Sequence.Consumer;
+
+                    processed++;
+                    bytesProcessed += msg.Size;
+
+                    yield return msg;
+                }
+            }
+            finally
             {
-                if (msg.Metadata is not { } metadata)
-                    continue;
+                var deleted = await TryDeleteConsumer(_fetchConsumerName, cancellationToken);
 
-                _fetchSeq = metadata.Sequence.Stream;
-                yield return msg;
+                if (deleted)
+                    _fetchConsumerName = string.Empty;
+            }
+
+            if (!mismatch)
+                yield break;
+
+            if (retry == _opts.MaxResetAttempts)
+            {
+                throw new NatsJSException("Maximum number of retry attempts reached.");
             }
-        }
-        finally
-        {
-            var deleted = await TryDeleteConsumer(_fetchConsumerName, cancellationToken);
 
-            if (deleted)
-                _fetchConsumerName = string.Empty;
+            await _context.Connection.Opts.BackoffWithJitterAsync(retry++, cancellationToken);
         }
     }
 

tests/NATS.Client.CoreUnit.Tests/NuidTests.cs

@@ -10,6 +10,7 @@ public class NuidTests
     private static readonly Regex NuidRegex = new("[A-z0-9]{22}");
 
     private readonly ITestOutputHelper _outputHelper;
+    private int _result;
 
     public NuidTests(ITestOutputHelper outputHelper)
     {
@@ -119,7 +120,6 @@ public void GetNextNuid_ContainsOnlyValidCharacters_Char()
     [Fact]
     public void GetNextNuid_PrefixRenewed_Char()
     {
-        var result = false;
         var firstNuid = new char[22];
         var secondNuid = new char[22];
 
@@ -129,15 +129,18 @@ public void GetNextNuid_PrefixRenewed_Char()
             var maxSequential = 839299365868340224ul - increment - 1;
             SetSequentialAndIncrement(maxSequential, increment);
 
-            result = Nuid.TryWriteNuid(firstNuid);
-            result &= Nuid.TryWriteNuid(secondNuid);
+            if (Nuid.TryWriteNuid(firstNuid))
+                Interlocked.Increment(ref _result);
+
+            if (Nuid.TryWriteNuid(secondNuid))
+                Interlocked.Increment(ref _result);
         });
 
         executionThread.Start();
         executionThread.Join(1_000);
 
         // Assert
-        Assert.True(result);
+        Assert.Equal(2, Interlocked.CompareExchange(ref _result, 0, 0));
         Assert.False(firstNuid.AsSpan(0, 12).SequenceEqual(secondNuid.AsSpan(0, 12)));
     }