feat: implement SubscriptionIndex + JetStreamMemStore cluster — 39 features verified

Add SubscriptionIndex factory methods, notification wrappers, and
ValidateMapping. Implement 24 MemStore methods (TTL, scheduling, SDM,
age-check, purge/compact/reset) with JetStream header helpers and
constants. Verified features: 987 → 1026.

dotnet/src/ZB.MOM.NatsNet.Server/JetStream/MemStore.cs

@@ -14,6 +14,7 @@
 // Adapted from server/memstore.go
 
 using System.Text;
+using ZB.MOM.NatsNet.Server.Internal;
 using ZB.MOM.NatsNet.Server.Internal.DataStructures;
 
 namespace ZB.MOM.NatsNet.Server;
@@ -57,6 +58,18 @@ public sealed class JetStreamMemStore : IStreamStore
     // Consumer count
     private int _consumers;
 
+    // TTL hash wheel (only created when cfg.AllowMsgTTL)
+    private HashWheel? _ttls;
+
+    // Message scheduling (only created when cfg.AllowMsgSchedules)
+    private MsgScheduling? _scheduling;
+
+    // Subject deletion metadata for cluster consensus
+    private StreamDeletionMeta _sdm = new();
+
+    // Guard against re-entrant age check
+    private bool _ageChkRun;
+
     // -----------------------------------------------------------------------
     // Constructor
     // -----------------------------------------------------------------------
@@ -83,6 +96,11 @@ public sealed class JetStreamMemStore : IStreamStore
             _state.LastSeq = cfg.FirstSeq - 1;
             _state.FirstSeq = cfg.FirstSeq;
         }
+
+        if (cfg.AllowMsgTTL)
+            _ttls = HashWheel.NewHashWheel();
+        if (cfg.AllowMsgSchedules)
+            _scheduling = new MsgScheduling(RunMsgScheduling);
     }
 
     // -----------------------------------------------------------------------
@@ -225,9 +243,52 @@ public sealed class JetStreamMemStore : IStreamStore
         EnforceMsgLimit();
         EnforceBytesLimit();
 
-        // Age check
-        if (_ageChk == null && _cfg.MaxAge != TimeSpan.Zero)
+        // Per-message TTL tracking.
+        if (_ttls != null && ttl > 0)
+        {
+            var expires = ts + (ttl * 1_000_000_000L);
+            _ttls.Add(seq, expires);
+        }
+
+        // Age check timer management.
+        if (_ttls != null && ttl > 0)
+        {
+            ResetAgeChk(0);
+        }
+        else if (_ageChk == null && (_cfg.MaxAge > TimeSpan.Zero || _ttls != null))
+        {
             StartAgeChk();
+        }
+
+        // Message scheduling.
+        if (_scheduling != null && hdr.Length > 0)
+        {
+            var (schedule, ok) = JetStreamHeaderHelpers.NextMessageSchedule(hdr, ts);
+            if (ok && schedule != default)
+            {
+                _scheduling.Add(seq, subject, schedule.Ticks * 100L);
+            }
+            else
+            {
+                _scheduling.RemoveSubject(subject);
+            }
+
+            // Check for a repeating schedule.
+            var scheduleNextSlice = NatsMessageHeaders.SliceHeader(NatsHeaderConstants.JsScheduleNext, hdr);
+            if (scheduleNextSlice != null && scheduleNextSlice.Value.Length > 0)
+            {
+                var scheduleNext = Encoding.ASCII.GetString(scheduleNextSlice.Value.Span);
+                if (scheduleNext != NatsHeaderConstants.JsScheduleNextPurge)
+                {
+                    var scheduler = JetStreamHeaderHelpers.GetMessageScheduler(hdr);
+                    if (DateTime.TryParse(scheduleNext, null, System.Globalization.DateTimeStyles.RoundtripKind, out var next)
+                        && !string.IsNullOrEmpty(scheduler))
+                    {
+                        _scheduling.Update(scheduler, next.ToUniversalTime().Ticks * 100L);
+                    }
+                }
+            }
+        }
     }
 
     /// <inheritdoc/>
@@ -562,6 +623,17 @@ public sealed class JetStreamMemStore : IStreamStore
         UpdateFirstSeq(seq);
         RemoveSeqPerSubject(sm.Subject, seq);
 
+        // Remove TTL entry from hash wheel if applicable.
+        if (_ttls != null && sm.Hdr.Length > 0)
+        {
+            var (ttl, err) = JetStreamHeaderHelpers.GetMessageTtl(sm.Hdr);
+            if (err == null && ttl > 0)
+            {
+                var expires = sm.Ts + (ttl * 1_000_000_000L);
+                _ttls.Remove(seq, expires);
+            }
+        }
+
         if (secure)
         {
             if (sm.Hdr.Length > 0)
@@ -1325,7 +1397,15 @@ public sealed class JetStreamMemStore : IStreamStore
     /// <inheritdoc/>
     public void ResetState()
     {
-        // For memory store, nothing to reset.
+        _mu.EnterWriteLock();
+        try
+        {
+            _scheduling?.ClearInflight();
+        }
+        finally
+        {
+            _mu.ExitWriteLock();
+        }
     }
 
     // -----------------------------------------------------------------------
@@ -1412,6 +1492,571 @@ public sealed class JetStreamMemStore : IStreamStore
         }
     }
 
+    // -----------------------------------------------------------------------
+    // Size helpers (static)
+    // -----------------------------------------------------------------------
+
+    /// <summary>
+    /// Computes raw message size from component lengths.
+    /// Mirrors Go <c>memStoreMsgSizeRaw</c>.
+    /// </summary>
+    internal static ulong MemStoreMsgSizeRaw(int slen, int hlen, int mlen)
+        => (ulong)(slen + hlen + mlen + 16);
+
+    /// <summary>
+    /// Computes message size from actual values.
+    /// Mirrors Go <c>memStoreMsgSize</c> (the package-level function).
+    /// </summary>
+    internal static ulong MemStoreMsgSize(string subj, byte[]? hdr, byte[]? msg)
+        => MemStoreMsgSizeRaw(subj.Length, hdr?.Length ?? 0, msg?.Length ?? 0);
+
+    // -----------------------------------------------------------------------
+    // Trivial helpers
+    // -----------------------------------------------------------------------
+
+    // Lock must be held.
+    private bool DeleteFirstMsg() => RemoveMsgLocked(_state.FirstSeq, false);
+
+    // Lock must be held.
+    private void DeleteFirstMsgOrPanic()
+    {
+        if (!DeleteFirstMsg())
+            throw new InvalidOperationException("jetstream memstore has inconsistent state, can't find first seq msg");
+    }
+
+    // Lock must be held.
+    private void CancelAgeChk()
+    {
+        if (_ageChk != null)
+        {
+            _ageChk.Dispose();
+            _ageChk = null;
+            _ageChkTime = 0;
+        }
+    }
+
+    /// <summary>
+    /// Returns true if a linear scan is preferable over subject tree lookup.
+    /// Mirrors Go <c>shouldLinearScan</c>.
+    /// </summary>
+    // Lock must be held.
+    private bool ShouldLinearScan(string filter, bool wc, ulong start)
+    {
+        const int LinearScanMaxFss = 256;
+        var isAll = filter == ">";
+        return isAll || 2 * (int)(_state.LastSeq - start) < _fss.Size() || (wc && _fss.Size() > LinearScanMaxFss);
+    }
+
+    /// <summary>
+    /// Returns true if the store is closed.
+    /// Mirrors Go <c>isClosed</c>.
+    /// </summary>
+    public bool IsClosed()
+    {
+        _mu.EnterReadLock();
+        try { return _msgs == null; }
+        finally { _mu.ExitReadLock(); }
+    }
+
+    /// <summary>
+    /// Checks if the filter represents all subjects (empty or "&gt;").
+    /// Mirrors Go <c>filterIsAll</c>.
+    /// </summary>
+    // Lock must be held.
+    private static bool FilterIsAll(string filter)
+        => string.IsNullOrEmpty(filter) || filter == ">";
+
+    // -----------------------------------------------------------------------
+    // Low-complexity helpers
+    // -----------------------------------------------------------------------
+
+    /// <summary>
+    /// Returns per-subject message totals matching the filter.
+    /// Mirrors Go <c>subjectsTotalsLocked</c>.
+    /// </summary>
+    // Lock must be held.
+    private Dictionary<string, ulong> SubjectsTotalsLocked(string filterSubject)
+    {
+        if (_fss.Size() == 0)
+            return new Dictionary<string, ulong>();
+
+        if (string.IsNullOrEmpty(filterSubject))
+            filterSubject = ">";
+
+        var isAll = filterSubject == ">";
+        var result = new Dictionary<string, ulong>();
+        _fss.Match(Encoding.UTF8.GetBytes(filterSubject), (subj, ss) =>
+        {
+            result[Encoding.UTF8.GetString(subj)] = ss.Msgs;
+            return true;
+        });
+        return result;
+    }
+
+    /// <summary>
+    /// Finds literal subject match sequence bounds.
+    /// Returns (first, last, true) if found, or (0, 0, false) if not.
+    /// Mirrors Go <c>nextLiteralMatchLocked</c>.
+    /// </summary>
+    // Lock must be held.
+    private (ulong First, ulong Last, bool Found) NextLiteralMatchLocked(string filter, ulong start)
+    {
+        var (ss, ok) = _fss.Find(Encoding.UTF8.GetBytes(filter));
+        if (!ok || ss == null)
+            return (0, 0, false);
+
+        RecalculateForSubj(filter, ss);
+        if (start > ss.Last)
+            return (0, 0, false);
+
+        return (Math.Max(start, ss.First), ss.Last, true);
+    }
+
+    /// <summary>
+    /// Finds wildcard subject match sequence bounds using MatchUntil.
+    /// Mirrors Go <c>nextWildcardMatchLocked</c>.
+    /// </summary>
+    // Lock must be held.
+    private (ulong First, ulong Last, bool Found) NextWildcardMatchLocked(string filter, ulong start)
+    {
+        bool found = false;
+        ulong first = _state.LastSeq, last = 0;
+
+        _fss.MatchUntil(Encoding.UTF8.GetBytes(filter), (subj, ss) =>
+        {
+            RecalculateForSubj(Encoding.UTF8.GetString(subj), ss);
+
+            if (start > ss.Last)
+                return true;
+
+            found = true;
+            if (ss.First < first)
+                first = ss.First;
+            if (ss.Last > last)
+                last = ss.Last;
+
+            return first > start;
+        });
+
+        if (!found)
+            return (0, 0, false);
+
+        return (Math.Max(first, start), last, true);
+    }
+
+    // -----------------------------------------------------------------------
+    // SDM methods
+    // -----------------------------------------------------------------------
+
+    /// <summary>
+    /// Determines whether this sequence/subject should be processed as a subject deletion marker.
+    /// Returns (isLast, shouldProcess).
+    /// Mirrors Go <c>shouldProcessSdmLocked</c>.
+    /// </summary>
+    // Lock must be held.
+    private (bool IsLast, bool ShouldProcess) ShouldProcessSdmLocked(ulong seq, string subj)
+    {
+        if (_sdm.TryGetPending(seq, out var p))
+        {
+            var elapsed = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L - p.Ts;
+            if (elapsed < 2_000_000_000L) // 2 seconds in nanoseconds
+                return (p.Last, false);
+
+            var last = p.Last;
+            if (last)
+            {
+                var msgs = SubjectsTotalsLocked(subj).GetValueOrDefault(subj, 0UL);
+                var numPending = _sdm.GetSubjectTotal(subj);
+                if (msgs > numPending)
+                    last = false;
+            }
+            _sdm.SetPending(seq, new SdmBySeq { Last = last, Ts = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L });
+            return (last, true);
+        }
+
+        var msgCount = SubjectsTotalsLocked(subj).GetValueOrDefault(subj, 0UL);
+        if (msgCount == 0)
+            return (false, true);
+
+        var pending = _sdm.GetSubjectTotal(subj);
+        var remaining = msgCount - pending;
+        return (_sdm.TrackPending(seq, subj, remaining == 1), true);
+    }
+
+    /// <summary>
+    /// Lock-wrapping version of ShouldProcessSdmLocked.
+    /// Mirrors Go <c>shouldProcessSdm</c>.
+    /// </summary>
+    public (bool IsLast, bool ShouldProcess) ShouldProcessSdm(ulong seq, string subj)
+    {
+        _mu.EnterWriteLock();
+        try { return ShouldProcessSdmLocked(seq, subj); }
+        finally { _mu.ExitWriteLock(); }
+    }
+
+    /// <summary>
+    /// Handles message removal: if SDM mode, builds a marker header and invokes _pmsgcb;
+    /// otherwise invokes _rmcb.
+    /// Mirrors Go <c>handleRemovalOrSdm</c>.
+    /// </summary>
+    public void HandleRemovalOrSdm(ulong seq, string subj, bool sdm, long sdmTtl)
+    {
+        if (sdm)
+        {
+            var hdr = Encoding.ASCII.GetBytes(
+                $"NATS/1.0\r\n{NatsHeaderConstants.JsMarkerReason}: {NatsHeaderConstants.JsMarkerReasonMaxAge}\r\n" +
+                $"{NatsHeaderConstants.JsMessageTtl}: {TimeSpan.FromSeconds(sdmTtl)}\r\n" +
+                $"{NatsHeaderConstants.JsMsgRollup}: {NatsHeaderConstants.JsMsgRollupSubject}\r\n\r\n");
+
+            // In Go this builds an inMsg and calls pmsgcb. We pass a synthetic StoreMsg.
+            var msg = new StoreMsg { Subject = subj, Hdr = hdr, Msg = Array.Empty<byte>(), Seq = 0, Ts = 0 };
+            _pmsgcb?.Invoke(msg);
+        }
+        else
+        {
+            _rmcb?.Invoke(seq);
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // Age/TTL methods
+    // -----------------------------------------------------------------------
+
+    /// <summary>
+    /// Resets or arms the age check timer based on TTL and MaxAge.
+    /// Mirrors Go <c>resetAgeChk</c>.
+    /// </summary>
+    // Lock must be held.
+    private void ResetAgeChk(long delta)
+    {
+        if (_ageChkRun)
+            return;
+
+        long next = long.MaxValue;
+        if (_ttls != null)
+            next = _ttls.GetNextExpiration(next);
+
+        if (_cfg.MaxAge <= TimeSpan.Zero && next == long.MaxValue)
+        {
+            CancelAgeChk();
+            return;
+        }
+
+        var fireIn = _cfg.MaxAge;
+
+        if (delta == 0 && _state.Msgs > 0)
+        {
+            var until = TimeSpan.FromSeconds(2);
+            if (fireIn == TimeSpan.Zero || until < fireIn)
+                fireIn = until;
+        }
+
+        if (next < long.MaxValue)
+        {
+            var nextTicks = DateTime.UnixEpoch.Ticks + next / 100L;
+            var nextUtc = new DateTime(Math.Max(nextTicks, DateTime.UnixEpoch.Ticks), DateTimeKind.Utc);
+            var until = nextUtc - DateTime.UtcNow;
+            if (fireIn == TimeSpan.Zero || until < fireIn)
+                fireIn = until;
+        }
+
+        if (delta > 0)
+        {
+            var deltaDur = TimeSpan.FromTicks(delta / 100L);
+            if (fireIn == TimeSpan.Zero || deltaDur < fireIn)
+                fireIn = deltaDur;
+        }
+
+        if (fireIn < TimeSpan.FromMilliseconds(250))
+            fireIn = TimeSpan.FromMilliseconds(250);
+
+        var expires = DateTime.UtcNow.Ticks + fireIn.Ticks;
+        if (_ageChkTime > 0 && expires > _ageChkTime)
+            return;
+
+        _ageChkTime = expires;
+        if (_ageChk != null)
+            _ageChk.Change(fireIn, Timeout.InfiniteTimeSpan);
+        else
+            _ageChk = new Timer(_ => ExpireMsgs(), null, fireIn, Timeout.InfiniteTimeSpan);
+    }
+
+    /// <summary>
+    /// Recovers TTL state from existing messages after restart.
+    /// Mirrors Go <c>recoverTTLState</c>.
+    /// </summary>
+    // Lock must be held.
+    private void RecoverTTLState()
+    {
+        _ttls = HashWheel.NewHashWheel();
+        if (_state.Msgs == 0)
+            return;
+
+        try
+        {
+            var smp = new StoreMsg();
+            var seq = _state.FirstSeq;
+            while (seq <= _state.LastSeq)
+            {
+                if (_msgs != null && _msgs.TryGetValue(seq, out var sm) && sm != null)
+                {
+                    if (sm.Hdr.Length > 0)
+                    {
+                        var (ttl, _) = JetStreamHeaderHelpers.GetMessageTtl(sm.Hdr);
+                        if (ttl > 0)
+                        {
+                            var expires = sm.Ts + (ttl * 1_000_000_000L);
+                            _ttls.Add(seq, expires);
+                        }
+                    }
+                }
+                seq++;
+            }
+        }
+        finally
+        {
+            ResetAgeChk(0);
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // Scheduling methods
+    // -----------------------------------------------------------------------
+
+    /// <summary>
+    /// Recovers message scheduling state from existing messages after restart.
+    /// Mirrors Go <c>recoverMsgSchedulingState</c>.
+    /// </summary>
+    // Lock must be held.
+    private void RecoverMsgSchedulingState()
+    {
+        _scheduling = new MsgScheduling(RunMsgScheduling);
+        if (_state.Msgs == 0)
+            return;
+
+        try
+        {
+            var seq = _state.FirstSeq;
+            while (seq <= _state.LastSeq)
+            {
+                if (_msgs != null && _msgs.TryGetValue(seq, out var sm) && sm != null)
+                {
+                    if (sm.Hdr.Length > 0)
+                    {
+                        var (schedule, ok) = JetStreamHeaderHelpers.NextMessageSchedule(sm.Hdr, sm.Ts);
+                        if (ok && schedule != default)
+                        {
+                            _scheduling.Init(seq, sm.Subject, schedule.Ticks * 100L);
+                        }
+                    }
+                }
+                seq++;
+            }
+        }
+        finally
+        {
+            _scheduling.ResetTimer();
+        }
+    }
+
+    /// <summary>
+    /// Runs through scheduled messages and fires callbacks.
+    /// Mirrors Go <c>runMsgScheduling</c>.
+    /// </summary>
+    private void RunMsgScheduling()
+    {
+        _mu.EnterWriteLock();
+        try
+        {
+            if (_scheduling == null)
+                return;
+            if (_pmsgcb == null)
+            {
+                _scheduling.ResetTimer();
+                return;
+            }
+
+            // TODO: Implement getScheduledMessages integration when MsgScheduling
+            // supports the full callback-based message loading pattern.
+            // For now, reset the timer so scheduling continues to fire.
+            _scheduling.ResetTimer();
+        }
+        finally
+        {
+            if (_mu.IsWriteLockHeld)
+                _mu.ExitWriteLock();
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // Reset / Purge Internal
+    // -----------------------------------------------------------------------
+
+    /// <summary>
+    /// Completely resets the store. Clears all messages, state, fss, dmap, and sdm.
+    /// Mirrors Go <c>reset</c>.
+    /// </summary>
+    public Exception? Reset()
+    {
+        _mu.EnterWriteLock();
+        ulong purged = 0;
+        ulong bytes = 0;
+        StorageUpdateHandler? cb;
+        try
+        {
+            cb = _scb;
+            if (cb != null && _msgs != null)
+            {
+                foreach (var sm in _msgs.Values)
+                {
+                    purged++;
+                    bytes += MsgSize(sm.Subject, sm.Hdr, sm.Msg);
+                }
+            }
+
+            _state.FirstSeq = 0;
+            _state.FirstTime = default;
+            _state.LastSeq = 0;
+            _state.LastTime = DateTime.UtcNow;
+            _state.Msgs = 0;
+            _state.Bytes = 0;
+            _msgs = new Dictionary<ulong, StoreMsg>();
+            _fss.Reset();
+            _dmap = new SequenceSet();
+            _sdm.Empty();
+        }
+        finally
+        {
+            _mu.ExitWriteLock();
+        }
+
+        cb?.Invoke(-(long)purged, -(long)bytes, 0, string.Empty);
+        return null;
+    }
+
+    /// <summary>
+    /// Internal purge with configurable first-sequence.
+    /// Mirrors Go <c>purge</c> (the internal version).
+    /// </summary>
+    // This is the internal purge used by cluster/raft — differs from the public Purge().
+    private (ulong Purged, Exception? Error) PurgeInternal(ulong fseq)
+    {
+        _mu.EnterWriteLock();
+        ulong purged;
+        long bytes;
+        StorageUpdateHandler? cb;
+        try
+        {
+            purged = (ulong)(_msgs?.Count ?? 0);
+            cb = _scb;
+            bytes = (long)_state.Bytes;
+
+            if (fseq == 0)
+                fseq = _state.LastSeq + 1;
+            else if (fseq < _state.LastSeq)
+            {
+                _mu.ExitWriteLock();
+                return (0, new InvalidOperationException("partial purges not supported on memory store"));
+            }
+
+            _state.FirstSeq = fseq;
+            _state.LastSeq = fseq - 1;
+            _state.FirstTime = default;
+            _state.Bytes = 0;
+            _state.Msgs = 0;
+            if (_msgs != null)
+                _msgs = new Dictionary<ulong, StoreMsg>();
+            _fss.Reset();
+            _dmap = new SequenceSet();
+            _sdm.Empty();
+        }
+        finally
+        {
+            if (_mu.IsWriteLockHeld)
+                _mu.ExitWriteLock();
+        }
+
+        cb?.Invoke(-(long)purged, -bytes, 0, string.Empty);
+        return (purged, null);
+    }
+
+    /// <summary>
+    /// Internal compact with SDM tracking.
+    /// Mirrors Go <c>compact</c> (the internal version).
+    /// </summary>
+    private (ulong Purged, Exception? Error) CompactInternal(ulong seq)
+    {
+        if (seq == 0)
+            return Purge();
+
+        ulong purged = 0;
+        ulong bytes = 0;
+
+        _mu.EnterWriteLock();
+        StorageUpdateHandler? cb;
+        try
+        {
+            if (_state.FirstSeq > seq)
+                return (0, null);
+
+            cb = _scb;
+            if (seq <= _state.LastSeq)
+            {
+                var fseq = _state.FirstSeq;
+                for (var s = seq; s <= _state.LastSeq; s++)
+                {
+                    if (_msgs != null && _msgs.TryGetValue(s, out var sm2) && sm2 != null)
+                    {
+                        _state.FirstSeq = s;
+                        _state.FirstTime = DateTimeOffset.FromUnixTimeMilliseconds(sm2.Ts / 1_000_000L).UtcDateTime;
+                        break;
+                    }
+                }
+                for (var s = seq - 1; s >= fseq; s--)
+                {
+                    if (_msgs != null && _msgs.TryGetValue(s, out var sm2) && sm2 != null)
+                    {
+                        bytes += MsgSize(sm2.Subject, sm2.Hdr, sm2.Msg);
+                        purged++;
+                        RemoveSeqPerSubject(sm2.Subject, s);
+                        _msgs.Remove(s);
+                    }
+                    else if (!_dmap.IsEmpty)
+                    {
+                        _dmap.Delete(s);
+                    }
+                    if (s == 0) break;
+                }
+                if (purged > _state.Msgs) purged = _state.Msgs;
+                _state.Msgs -= purged;
+                if (bytes > _state.Bytes) bytes = _state.Bytes;
+                _state.Bytes -= bytes;
+            }
+            else
+            {
+                purged = (ulong)(_msgs?.Count ?? 0);
+                bytes = _state.Bytes;
+                _state.Bytes = 0;
+                _state.Msgs = 0;
+                _state.FirstSeq = seq;
+                _state.FirstTime = default;
+                _state.LastSeq = seq - 1;
+                _msgs = new Dictionary<ulong, StoreMsg>();
+                _fss.Reset();
+                _dmap = new SequenceSet();
+                _sdm.Empty();
+            }
+        }
+        finally
+        {
+            if (_mu.IsWriteLockHeld)
+                _mu.ExitWriteLock();
+        }
+
+        cb?.Invoke(-(long)purged, -(long)bytes, 0, string.Empty);
+        return (purged, null);
+    }
+
     // -----------------------------------------------------------------------
     // Private helpers
     // -----------------------------------------------------------------------