feat(consumers): add NAK/TERM/PROGRESS ack types with backoff (Go parity)

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

@@ -1,9 +1,21 @@
+// Go: consumer.go (processAckMsg, processNak, processTerm, processAckProgress)
 namespace NATS.Server.JetStream.Consumers;
 
 public sealed class AckProcessor
 {
+    // Go: consumer.go — ackTerminatedFlag marks sequences that must not be redelivered
+    private readonly HashSet<ulong> _terminated = new();
     private readonly Dictionary<ulong, PendingState> _pending = new();
+    private readonly int[]? _backoffMs;
+    private int _ackWaitMs;
+
     public ulong AckFloor { get; private set; }
+    public int TerminatedCount { get; private set; }
+
+    public AckProcessor(int[]? backoffMs = null)
+    {
+        _backoffMs = backoffMs;
+    }
 
     public void Register(ulong sequence, int ackWaitMs)
     {
@@ -13,6 +25,8 @@ public sealed class AckProcessor
         if (_pending.ContainsKey(sequence))
             return;
 
+        _ackWaitMs = ackWaitMs;
+
         _pending[sequence] = new PendingState
         {
             DeadlineUtc = DateTime.UtcNow.AddMilliseconds(Math.Max(ackWaitMs, 1)),
@@ -37,6 +51,120 @@ public sealed class AckProcessor
         return false;
     }
 
+    // Go: consumer.go:2550 (processAck)
+    // Dispatches to the appropriate ack handler based on ack type prefix.
+    // Empty or "+ACK" → ack single; "-NAK" → schedule redelivery; "+TERM" → terminate; "+WPI" → progress reset.
+    public void ProcessAck(ulong seq, ReadOnlySpan<byte> payload)
+    {
+        if (payload.IsEmpty || payload.SequenceEqual("+ACK"u8))
+        {
+            AckSequence(seq);
+            return;
+        }
+
+        if (payload.StartsWith("-NAK"u8))
+        {
+            // Go: consumer.go — parseNak extracts optional delay from "-NAK {delay}"
+            var delayMs = 0;
+            var rest = payload["-NAK"u8.Length..];
+            if (!rest.IsEmpty && rest[0] == (byte)' ')
+            {
+                var delaySpan = rest[1..];
+                if (TryParseInt(delaySpan, out var parsed))
+                    delayMs = parsed;
+            }
+            ProcessNak(seq, delayMs);
+            return;
+        }
+
+        if (payload.StartsWith("+TERM"u8))
+        {
+            ProcessTerm(seq);
+            return;
+        }
+
+        if (payload.StartsWith("+WPI"u8))
+        {
+            ProcessProgress(seq);
+            return;
+        }
+
+        // Unknown ack type — treat as plain ack per Go behavior
+        AckSequence(seq);
+    }
+
+    // Go: consumer.go — processAck for "+ACK": removes from pending and advances AckFloor when contiguous
+    public void AckSequence(ulong seq)
+    {
+        _pending.Remove(seq);
+        _terminated.Remove(seq);
+
+        // Advance floor while the next-in-order sequences are no longer pending
+        if (seq == AckFloor + 1)
+        {
+            AckFloor = seq;
+            while (_pending.Count > 0)
+            {
+                var next = AckFloor + 1;
+                if (_pending.ContainsKey(next))
+                    break;
+                // Only advance if next is definitely below any pending sequence
+                // Stop when we hit a gap or run out of sequences to check
+                if (!HasSequenceBelow(next))
+                    break;
+                AckFloor = next;
+            }
+        }
+    }
+
+    // Go: consumer.go — processNak: schedules redelivery with optional explicit delay or backoff array
+    public void ProcessNak(ulong seq, int delayMs = 0)
+    {
+        if (_terminated.Contains(seq))
+            return;
+
+        if (!_pending.TryGetValue(seq, out var state))
+            return;
+
+        int effectiveDelay;
+        if (delayMs > 0)
+        {
+            effectiveDelay = delayMs;
+        }
+        else if (_backoffMs is { Length: > 0 })
+        {
+            // Go: consumer.go — backoff array clamps at last entry for high delivery counts
+            var idx = Math.Min(state.Deliveries - 1, _backoffMs.Length - 1);
+            effectiveDelay = _backoffMs[idx];
+        }
+        else
+        {
+            effectiveDelay = Math.Max(_ackWaitMs, 1);
+        }
+
+        ScheduleRedelivery(seq, effectiveDelay);
+    }
+
+    // Go: consumer.go — processTerm: removes from pending permanently; sequence is never redelivered
+    public void ProcessTerm(ulong seq)
+    {
+        if (_pending.Remove(seq))
+        {
+            _terminated.Add(seq);
+            TerminatedCount++;
+        }
+    }
+
+    // Go: consumer.go — processAckProgress (+WPI): resets ack deadline to original ackWait without bumping delivery count
+    public void ProcessProgress(ulong seq)
+    {
+        if (!_pending.TryGetValue(seq, out var state))
+            return;
+
+        state.DeadlineUtc = DateTime.UtcNow.AddMilliseconds(Math.Max(_ackWaitMs, 1));
+        _pending[seq] = state;
+    }
+
     public void ScheduleRedelivery(ulong sequence, int delayMs)
     {
         if (!_pending.TryGetValue(sequence, out var state))
@@ -64,6 +192,31 @@ public sealed class AckProcessor
             AckFloor = sequence;
     }
 
+    private bool HasSequenceBelow(ulong upTo)
+    {
+        foreach (var key in _pending.Keys)
+        {
+            if (key < upTo)
+                return true;
+        }
+        return false;
+    }
+
+    private static bool TryParseInt(ReadOnlySpan<byte> span, out int value)
+    {
+        value = 0;
+        if (span.IsEmpty)
+            return false;
+
+        foreach (var b in span)
+        {
+            if (b < (byte)'0' || b > (byte)'9')
+                return false;
+            value = value * 10 + (b - '0');
+        }
+        return true;
+    }
+
     private sealed class PendingState
     {
         public DateTime DeadlineUtc { get; set; }