feat: add max delivery enforcement with advisory generation (Gap 3.9)

Adds MaxDeliver property, exceeded sequence tracking, DrainExceeded,
DeliveryExceededPolicy enum, and ScheduleRedelivery enforcement to
AckProcessor; 10 new tests in MaxDeliveriesTests.cs.

src/NATS.Server/JetStream/Consumers/AckProcessor.cs

@@ -4,6 +4,18 @@ namespace NATS.Server.JetStream.Consumers;
 // Go: consumer.go:2550 — ack type prefix constants (+ACK, -NAK, +TERM, +WPI)
 public enum AckType { Ack, Nak, Term, Progress, Unknown }
 
+/// <summary>
+/// Policy applied when a sequence exceeds its max delivery count.
+/// Go reference: consumer.go maxDeliver config.
+/// </summary>
+public enum DeliveryExceededPolicy
+{
+    /// <summary>Drop the message — no further action.</summary>
+    Drop,
+    /// <summary>Move to a dead-letter queue subject.</summary>
+    DeadLetter,
+}
+
 public sealed class AckProcessor
 {
     // Go: consumer.go — ackTerminatedFlag marks sequences that must not be redelivered
@@ -19,9 +31,35 @@ public sealed class AckProcessor
     // Stores deliver subjects keyed by sequence for tracker-based registrations
     private readonly Dictionary<ulong, string>? _deliverSubjects;
 
+    // Go: consumer.go — maxDeliver config; sequences that exceeded the limit
+    private int _maxDeliver;
+    private readonly List<ulong> _exceededSequences = new();
+
     public ulong AckFloor { get; private set; }
     public int TerminatedCount { get; private set; }
 
+    /// <summary>
+    /// Sets the max delivery limit. 0 or negative means unlimited.
+    /// Go reference: consumer.go maxDeliver config.
+    /// </summary>
+    public int MaxDeliver
+    {
+        get => _maxDeliver;
+        set => _maxDeliver = Math.Max(value, 0);
+    }
+
+    /// <summary>Number of sequences that exceeded max deliveries.</summary>
+    public int ExceededCount => _exceededSequences.Count;
+
+    /// <summary>Returns sequences that exceeded max deliveries since last drain.</summary>
+    public IReadOnlyList<ulong> GetExceededSequences() => _exceededSequences.AsReadOnly();
+
+    /// <summary>Clears the exceeded sequences list (after advisories are sent).</summary>
+    public void DrainExceeded() => _exceededSequences.Clear();
+
+    /// <summary>Policy applied when a sequence exceeds its max delivery count.</summary>
+    public DeliveryExceededPolicy ExceededPolicy { get; set; } = DeliveryExceededPolicy.Drop;
+
     public AckProcessor(int[]? backoffMs = null)
     {
         _backoffMs = backoffMs;
@@ -236,6 +274,17 @@ public sealed class AckProcessor
             return;
 
         state.Deliveries++;
+
+        // Go: consumer.go — check maxDeliver; Deliveries > maxDeliver means the limit is exceeded
+        if (_maxDeliver > 0 && state.Deliveries > _maxDeliver)
+        {
+            _exceededSequences.Add(sequence);
+            _pending.Remove(sequence);
+            _terminated.Add(sequence);
+            TerminatedCount++;
+            return;
+        }
+
         state.DeadlineUtc = DateTime.UtcNow.AddMilliseconds(Math.Max(delayMs, 1));
         _pending[sequence] = state;
     }