feat: implement strict retention runtime parity for jetstream

src/NATS.Server/JetStream/ConsumerManager.cs

@@ -12,6 +12,7 @@ public sealed class ConsumerManager
 {
     private readonly JetStreamMetaGroup? _metaGroup;
     private readonly ConcurrentDictionary<(string Stream, string Name), ConsumerHandle> _consumers = new();
+    private readonly ConcurrentDictionary<string, ulong> _ackFloors = new(StringComparer.Ordinal);
     private readonly PullConsumerEngine _pullConsumerEngine = new();
     private readonly PushConsumerEngine _pushConsumerEngine = new();
 
@@ -130,6 +131,7 @@ public sealed class ConsumerManager
             return false;
 
         handle.AckProcessor.AckAll(sequence);
+        _ackFloors.AddOrUpdate(stream, _ => sequence, (_, existing) => Math.Max(existing, sequence));
         return true;
     }
 
@@ -180,6 +182,9 @@ public sealed class ConsumerManager
 
         return true;
     }
+
+    internal ulong GetAckFloor(string stream)
+        => _ackFloors.TryGetValue(stream, out var ackFloor) ? ackFloor : 0;
 }
 
 public sealed record ConsumerHandle(string Stream, ConsumerConfig Config)

src/NATS.Server/JetStream/StreamManager.cs

@@ -14,6 +14,7 @@ namespace NATS.Server.JetStream;
 public sealed class StreamManager
 {
     private readonly Account? _account;
+    private readonly ConsumerManager? _consumerManager;
     private readonly JetStreamMetaGroup? _metaGroup;
     private readonly ConcurrentDictionary<string, StreamHandle> _streams =
         new(StringComparer.Ordinal);
@@ -25,10 +26,11 @@ public sealed class StreamManager
         new(StringComparer.Ordinal);
     private readonly StreamSnapshotService _snapshotService = new();
 
-    public StreamManager(JetStreamMetaGroup? metaGroup = null, Account? account = null)
+    public StreamManager(JetStreamMetaGroup? metaGroup = null, Account? account = null, ConsumerManager? consumerManager = null)
     {
         _metaGroup = metaGroup;
         _account = account;
+        _consumerManager = consumerManager;
     }
 
     public IReadOnlyCollection<string> StreamNames => _streams.Keys.ToArray();
@@ -261,7 +263,7 @@ public sealed class StreamManager
         };
     }
 
-    private static void EnforceRuntimePolicies(StreamHandle stream, DateTime nowUtc)
+    private void EnforceRuntimePolicies(StreamHandle stream, DateTime nowUtc)
     {
         switch (stream.Config.Retention)
         {
@@ -284,11 +286,23 @@ public sealed class StreamManager
         PruneExpiredMessages(stream, nowUtc);
     }
 
-    private static void ApplyWorkQueueRetention(StreamHandle stream, DateTime nowUtc)
+    private void ApplyWorkQueueRetention(StreamHandle stream, DateTime nowUtc)
     {
-        // WorkQueue keeps one-consumer processing semantics; current parity baseline
-        // applies the same bounded retention guards used by limits retention.
         ApplyLimitsRetention(stream, nowUtc);
+
+        if (_consumerManager == null)
+            return;
+
+        var ackFloor = _consumerManager.GetAckFloor(stream.Config.Name);
+        if (ackFloor == 0)
+            return;
+
+        var messages = stream.Store.ListAsync(default).GetAwaiter().GetResult();
+        foreach (var message in messages)
+        {
+            if (message.Sequence <= ackFloor)
+                stream.Store.RemoveAsync(message.Sequence, default).GetAwaiter().GetResult();
+        }
     }
 
     private static void ApplyInterestRetention(StreamHandle stream, DateTime nowUtc)

src/NATS.Server/NatsServer.cs

@@ -408,8 +408,8 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
 
         if (options.JetStream != null)
         {
-            _jetStreamStreamManager = new StreamManager();
             _jetStreamConsumerManager = new ConsumerManager();
+            _jetStreamStreamManager = new StreamManager(consumerManager: _jetStreamConsumerManager);
             var jsClientId = Interlocked.Increment(ref _nextClientId);
             _jetStreamInternalClient = new InternalClient(jsClientId, ClientKind.JetStream, _systemAccount);
             _jetStreamService = new JetStreamService(options.JetStream, _jetStreamInternalClient);