chore(batch32): reconcile porting.db after merge

dotnet/src/ZB.MOM.NatsNet.Server/Accounts/Account.JetStream.cs

@@ -127,6 +127,57 @@ public sealed partial class Account
         return (server, jsa, null);
     }
 
+    internal (NatsServer? Server, JetStream? JetStream, JsAccount? JetStreamAccount) GetJetStreamFromAccount()
+    {
+        _mu.EnterReadLock();
+        var jetStreamAccount = JetStream;
+        _mu.ExitReadLock();
+        if (jetStreamAccount == null)
+            return (null, null, null);
+
+        jetStreamAccount.Lock.EnterReadLock();
+        var jetStream = jetStreamAccount.Js as JetStream;
+        jetStreamAccount.Lock.ExitReadLock();
+        if (jetStream == null)
+            return (null, null, null);
+
+        return (jetStream.Server as NatsServer, jetStream, jetStreamAccount);
+    }
+
+    internal bool JetStreamIsStreamLeader(string stream)
+    {
+        var (server, jetStream, _) = GetJetStreamFromAccount();
+        if (server == null || jetStream == null)
+            return false;
+
+        jetStream.Lock.EnterReadLock();
+        try
+        {
+            return (jetStream.Cluster as JetStreamCluster)?.IsStreamLeader(Name, stream) == true;
+        }
+        finally
+        {
+            jetStream.Lock.ExitReadLock();
+        }
+    }
+
+    internal bool JetStreamIsConsumerLeader(string stream, string consumer)
+    {
+        var (server, jetStream, _) = GetJetStreamFromAccount();
+        if (server == null || jetStream == null)
+            return false;
+
+        jetStream.Lock.EnterReadLock();
+        try
+        {
+            return (jetStream.Cluster as JetStreamCluster)?.IsConsumerLeader(Name, stream, consumer) == true;
+        }
+        finally
+        {
+            jetStream.Lock.ExitReadLock();
+        }
+    }
+
     internal (bool MaxBytesRequired, long MaxStreamBytes) MaxBytesLimits(StreamConfig? cfg)
     {
         _mu.EnterReadLock();

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

@@ -90,6 +90,252 @@ internal sealed class JetStreamCluster
 
     /// <summary>Duration of last meta-snapshot (nanoseconds).</summary>
     public long LastMetaSnapDuration { get; set; }
+
+    internal static UnsupportedStreamAssignment NewUnsupportedStreamAssignment(
+        NatsServer server,
+        StreamAssignment assignment,
+        Exception? error)
+    {
+        var reason = "stopped";
+        if (error != null)
+        {
+            var msg = error.Message;
+            reason = msg.StartsWith("json:", StringComparison.Ordinal)
+                ? $"unsupported - config error: {msg["json:".Length..].TrimStart()}"
+                : $"stopped - {msg}";
+        }
+        else if (assignment.Config != null && !JetStreamVersioning.SupportsRequiredApiLevel(assignment.Config.Metadata))
+        {
+            var required = JetStreamVersioning.GetRequiredApiLevel(assignment.Config.Metadata);
+            if (!string.IsNullOrEmpty(required))
+                reason = $"unsupported - required API level: {required}, current API level: {JetStreamVersioning.JsApiLevel}";
+        }
+
+        var config = assignment.Config ?? new StreamConfig { Name = assignment.Subject ?? string.Empty };
+        return new UnsupportedStreamAssignment
+        {
+            Reason = reason,
+            Info = new StreamInfo
+            {
+                Created = assignment.Created,
+                Config = JetStreamVersioning.SetDynamicStreamMetadata(config),
+            },
+        };
+    }
+
+    internal static UnsupportedConsumerAssignment NewUnsupportedConsumerAssignment(
+        ConsumerAssignment assignment,
+        Exception? error)
+    {
+        var reason = "stopped";
+        if (error != null)
+        {
+            var msg = error.Message;
+            reason = msg.StartsWith("json:", StringComparison.Ordinal)
+                ? $"unsupported - config error: {msg["json:".Length..].TrimStart()}"
+                : $"stopped - {msg}";
+        }
+        else if (assignment.Config != null && !JetStreamVersioning.SupportsRequiredApiLevel(assignment.Config.Metadata))
+        {
+            var required = JetStreamVersioning.GetRequiredApiLevel(assignment.Config.Metadata);
+            if (!string.IsNullOrEmpty(required))
+                reason = $"unsupported - required API level: {required}, current API level: {JetStreamVersioning.JsApiLevel}";
+        }
+
+        var config = assignment.Config ?? new ConsumerConfig { Name = assignment.Name };
+        return new UnsupportedConsumerAssignment
+        {
+            Reason = reason,
+            Info = new ConsumerInfo
+            {
+                Stream = assignment.Stream,
+                Name = assignment.Name,
+                Created = assignment.Created,
+                Config = JetStreamVersioning.SetDynamicConsumerMetadata(config),
+                TimeStamp = DateTime.UtcNow,
+            },
+        };
+    }
+
+    internal bool IsLeader()
+    {
+        if (this == null)
+            return true;
+
+        return Meta != null && Meta.Leader();
+    }
+
+    internal bool IsStreamCurrent(string account, string stream)
+    {
+        if (Meta == null)
+            return false;
+
+        if (!Streams.TryGetValue(account, out var accountAssignments))
+            return false;
+        if (!accountAssignments.TryGetValue(stream, out var assignment))
+            return false;
+        if (assignment.Group == null)
+            return false;
+
+        return assignment.Group.Node == null || assignment.Group.Node.Current();
+    }
+
+    internal bool IsStreamAssigned(Account account, string stream)
+    {
+        if (Meta == null)
+            return false;
+
+        if (!Streams.TryGetValue(account.Name, out var accountAssignments))
+            return false;
+        if (!accountAssignments.TryGetValue(stream, out var assignment))
+            return false;
+
+        var peers = assignment.Group?.Peers ?? [];
+        return peers.Contains(Meta.ID(), StringComparer.Ordinal);
+    }
+
+    internal bool IsStreamLeader(string account, string stream)
+    {
+        if (Meta == null)
+            return false;
+
+        if (!Streams.TryGetValue(account, out var accountAssignments))
+            return false;
+        if (!accountAssignments.TryGetValue(stream, out var assignment))
+            return false;
+
+        var group = assignment.Group;
+        if (group == null)
+            return false;
+
+        var ourId = Meta.ID();
+        if (!group.Peers.Contains(ourId, StringComparer.Ordinal))
+            return false;
+
+        return group.Peers.Length == 1 || (group.Node != null && group.Node.Leader());
+    }
+
+    internal bool IsConsumerLeader(string account, string stream, string consumer)
+    {
+        if (Meta == null)
+            return false;
+
+        if (!Streams.TryGetValue(account, out var accountAssignments))
+            return false;
+        if (!accountAssignments.TryGetValue(stream, out var assignment))
+            return false;
+        if (assignment.Consumers == null || !assignment.Consumers.TryGetValue(consumer, out var consumerAssignment))
+            return false;
+
+        var group = consumerAssignment.Group;
+        if (group == null)
+            return false;
+
+        var ourId = Meta.ID();
+        if (!group.Peers.Contains(ourId, StringComparer.Ordinal))
+            return false;
+
+        return group.Peers.Length == 1 || (group.Node != null && group.Node.Leader());
+    }
+
+    internal void TrackInflightStreamProposal(string accountName, StreamAssignment assignment, bool deleted)
+    {
+        if (!InflightStreams.TryGetValue(accountName, out var streams))
+        {
+            streams = new Dictionary<string, InflightStreamInfo>(StringComparer.Ordinal);
+            InflightStreams[accountName] = streams;
+        }
+
+        var streamName = assignment.Config?.Name ?? assignment.Subject ?? string.Empty;
+        if (streams.TryGetValue(streamName, out var inflight))
+        {
+            inflight.Ops++;
+            inflight.Deleted = deleted;
+            inflight.Assignment = assignment;
+            return;
+        }
+
+        streams[streamName] = new InflightStreamInfo
+        {
+            Ops = 1,
+            Deleted = deleted,
+            Assignment = assignment,
+        };
+    }
+
+    internal void RemoveInflightStreamProposal(string accountName, string streamName)
+    {
+        if (!InflightStreams.TryGetValue(accountName, out var streams))
+            return;
+        if (!streams.TryGetValue(streamName, out var inflight))
+            return;
+
+        if (inflight.Ops > 1)
+        {
+            inflight.Ops--;
+            return;
+        }
+
+        streams.Remove(streamName);
+        if (streams.Count == 0)
+            InflightStreams.Remove(accountName);
+    }
+
+    internal void TrackInflightConsumerProposal(
+        string accountName,
+        string streamName,
+        ConsumerAssignment assignment,
+        bool deleted)
+    {
+        if (!InflightConsumers.TryGetValue(accountName, out var streams))
+        {
+            streams = new Dictionary<string, Dictionary<string, InflightConsumerInfo>>(StringComparer.Ordinal);
+            InflightConsumers[accountName] = streams;
+        }
+
+        if (!streams.TryGetValue(streamName, out var consumers))
+        {
+            consumers = new Dictionary<string, InflightConsumerInfo>(StringComparer.Ordinal);
+            streams[streamName] = consumers;
+        }
+
+        if (consumers.TryGetValue(assignment.Name, out var inflight))
+        {
+            inflight.Ops++;
+            inflight.Deleted = deleted;
+            inflight.Assignment = assignment;
+            return;
+        }
+
+        consumers[assignment.Name] = new InflightConsumerInfo
+        {
+            Ops = 1,
+            Deleted = deleted,
+            Assignment = assignment,
+        };
+    }
+
+    internal void RemoveInflightConsumerProposal(string accountName, string streamName, string consumerName)
+    {
+        if (!InflightConsumers.TryGetValue(accountName, out var streams))
+            return;
+        if (!streams.TryGetValue(streamName, out var consumers))
+            return;
+        if (!consumers.TryGetValue(consumerName, out var inflight))
+            return;
+
+        if (inflight.Ops > 1)
+        {
+            inflight.Ops--;
+            return;
+        }
+
+        consumers.Remove(consumerName);
+        if (consumers.Count == 0)
+            streams.Remove(streamName);
+        if (streams.Count == 0)
+            InflightConsumers.Remove(accountName);
+    }
 }
 
 // ============================================================================
@@ -255,6 +501,28 @@ internal sealed class UnsupportedStreamAssignment
     public object? SysClient   { get; set; }
     /// <summary>Info subscription (object to avoid session dep).</summary>
     public object? InfoSub     { get; set; }
+
+    internal void SetupInfoSub(NatsServer server, StreamAssignment assignment)
+    {
+        if (InfoSub != null)
+            return;
+
+        SysClient = server.CreateInternalJetStreamClient();
+        InfoSub = $"{assignment.Client?.ServiceAccount() ?? string.Empty}:{assignment.Config?.Name ?? string.Empty}";
+    }
+
+    internal StreamInfoClusterResponse HandleClusterStreamInfoRequest() =>
+        new()
+        {
+            OfflineReason = Reason,
+            StreamInfo = Info,
+        };
+
+    internal void CloseInfoSub()
+    {
+        InfoSub = null;
+        SysClient = null;
+    }
 }
 
 // ============================================================================
@@ -301,6 +569,28 @@ internal sealed class UnsupportedConsumerAssignment
     public object? SysClient      { get; set; }
     /// <summary>Info subscription (object to avoid session dep).</summary>
     public object? InfoSub        { get; set; }
+
+    internal void SetupInfoSub(NatsServer server, ConsumerAssignment assignment)
+    {
+        if (InfoSub != null)
+            return;
+
+        SysClient = server.CreateInternalJetStreamClient();
+        InfoSub = $"{assignment.Client?.ServiceAccount() ?? string.Empty}:{assignment.Stream}:{assignment.Name}";
+    }
+
+    internal ConsumerInfoClusterResponse HandleClusterConsumerInfoRequest() =>
+        new()
+        {
+            OfflineReason = Reason,
+            ConsumerInfo = Info,
+        };
+
+    internal void CloseInfoSub()
+    {
+        InfoSub = null;
+        SysClient = null;
+    }
 }
 
 // ============================================================================
@@ -374,6 +664,85 @@ internal sealed class RecoveryUpdates
     public Dictionary<string, StreamAssignment>                          AddStreams      { get; set; } = new();
     public Dictionary<string, StreamAssignment>                          UpdateStreams   { get; set; } = new();
     public Dictionary<string, Dictionary<string, ConsumerAssignment>>   UpdateConsumers { get; set; } = new();
+
+    internal void RemoveStream(StreamAssignment assignment)
+    {
+        var key = StreamRecoveryKey(assignment);
+        RemoveStreams[key] = assignment;
+        AddStreams.Remove(key);
+        UpdateStreams.Remove(key);
+        UpdateConsumers.Remove(key);
+        RemoveConsumers.Remove(key);
+    }
+
+    internal void AddStream(StreamAssignment assignment)
+    {
+        var key = StreamRecoveryKey(assignment);
+        AddStreams[key] = assignment;
+    }
+
+    internal void UpdateStream(StreamAssignment assignment)
+    {
+        var key = StreamRecoveryKey(assignment);
+        UpdateStreams[key] = assignment;
+    }
+
+    internal void RemoveConsumer(ConsumerAssignment assignment)
+    {
+        var streamKey = ConsumerStreamRecoveryKey(assignment);
+        var consumerKey = ConsumerRecoveryKey(assignment);
+
+        if (!RemoveConsumers.TryGetValue(streamKey, out var consumers))
+        {
+            consumers = new Dictionary<string, ConsumerAssignment>(StringComparer.Ordinal);
+            RemoveConsumers[streamKey] = consumers;
+        }
+
+        consumers[consumerKey] = assignment;
+        if (UpdateConsumers.TryGetValue(streamKey, out var updates))
+            updates.Remove(consumerKey);
+    }
+
+    internal void AddOrUpdateConsumer(ConsumerAssignment assignment)
+    {
+        var streamKey = ConsumerStreamRecoveryKey(assignment);
+        var consumerKey = ConsumerRecoveryKey(assignment);
+
+        if (!UpdateConsumers.TryGetValue(streamKey, out var consumers))
+        {
+            consumers = new Dictionary<string, ConsumerAssignment>(StringComparer.Ordinal);
+            UpdateConsumers[streamKey] = consumers;
+        }
+
+        consumers[consumerKey] = assignment;
+    }
+
+    private static string StreamRecoveryKey(StreamAssignment assignment) =>
+        $"{assignment.Client?.ServiceAccount() ?? string.Empty}:{assignment.Config?.Name ?? assignment.Subject ?? string.Empty}";
+
+    private static string ConsumerStreamRecoveryKey(ConsumerAssignment assignment) =>
+        $"{assignment.Client?.ServiceAccount() ?? string.Empty}:{assignment.Stream}";
+
+    private static string ConsumerRecoveryKey(ConsumerAssignment assignment) =>
+        $"{assignment.Stream}:{assignment.Name}";
+}
+
+internal sealed class StreamInfoClusterResponse
+{
+    [JsonPropertyName("offline_reason")]
+    public string? OfflineReason { get; set; }
+
+    [JsonPropertyName("stream_info")]
+    public StreamInfo? StreamInfo { get; set; }
+}
+
+internal sealed class ConsumerInfoClusterResponse
+{
+    [JsonPropertyName("offline_reason")]
+    public string? OfflineReason { get; set; }
+
+    [JsonPropertyName("consumer_info")]
+    public ConsumerInfo? ConsumerInfo { get; set; }
 }
 
 // ============================================================================

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

@@ -324,6 +324,297 @@ internal sealed class JetStreamEngine(JetStream state)
             Interlocked.Add(ref _state.StoreReserved, -cfg.MaxBytes);
     }
 
+    internal bool IsStreamHealthy(Account? account, StreamAssignment? assignment)
+    {
+        if (assignment?.Unsupported != null)
+            return true;
+        if (_state.Cluster is not JetStreamCluster)
+            return true;
+        if (assignment?.Group == null || assignment.Config == null || account == null)
+            return false;
+
+        var (stream, _) = account.LookupStream(assignment.Config.Name);
+        if (stream == null)
+            return false;
+
+        if (assignment.Config.Replicas <= 1)
+            return true;
+
+        var node = assignment.Group.Node;
+        return node != null && node.Healthy();
+    }
+
+    internal bool IsConsumerHealthy(NatsStream? stream, string consumer, ConsumerAssignment? assignment)
+    {
+        if (assignment?.Unsupported != null)
+            return true;
+        if (_state.Cluster is not JetStreamCluster)
+            return true;
+        if (stream == null || assignment?.Group == null)
+            return false;
+
+        if (assignment.Group.Peers.Length <= 1)
+            return true;
+
+        return assignment.Group.Node?.Healthy() == true;
+    }
+
+    internal bool SubjectsOverlap(string account, IEnumerable<string> subjects, StreamAssignment? ownAssignment = null)
+    {
+        if (_state.Cluster is not JetStreamCluster cluster)
+            return false;
+
+        var targetSubjects = subjects.ToArray();
+        if (!cluster.Streams.TryGetValue(account, out var accountAssignments))
+            return false;
+
+        foreach (var assignment in accountAssignments.Values)
+        {
+            if (ownAssignment != null &&
+                assignment.Config?.Name != null &&
+                assignment.Config.Name == ownAssignment.Config?.Name)
+            {
+                continue;
+            }
+
+            var existingSubjects = assignment.Config?.Subjects ?? [];
+            foreach (var existing in existingSubjects)
+            {
+                foreach (var target in targetSubjects)
+                {
+                    if (Internal.DataStructures.SubscriptionIndex.SubjectsCollide(target, existing))
+                        return true;
+                }
+            }
+        }
+
+        return false;
+    }
+
+    internal bool IsClustered() => _state.Cluster is JetStreamCluster;
+
+    internal bool IsClusteredNoLock() => Interlocked.CompareExchange(ref _state.Clustered, 0, 0) == 1;
+
+    internal Exception? SetupMetaGroup()
+    {
+        var server = Server();
+        if (server == null)
+            return new InvalidOperationException("jetstream server unavailable");
+
+        _state.Lock.EnterWriteLock();
+        try
+        {
+            if (_state.Cluster is JetStreamCluster)
+            {
+                Interlocked.Exchange(ref _state.Clustered, 1);
+                return null;
+            }
+
+            var systemAccount = server.SystemAccount();
+            if (systemAccount == null)
+                return ServerErrors.ErrNoSysAccount;
+
+            var storeDir = Path.Combine(_state.Config.StoreDir, systemAccount.Name, "_js_", "_meta_");
+            Directory.CreateDirectory(storeDir);
+
+            var raftConfig = new RaftConfig
+            {
+                Name = "_meta_",
+                Store = storeDir,
+                Recovering = true,
+            };
+
+            var (meta, error) = server.StartRaftNode(systemAccount.Name, raftConfig);
+            if (error != null)
+                return error;
+
+            _state.Cluster = new JetStreamCluster
+            {
+                Meta = meta,
+                Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>(StringComparer.Ordinal),
+                Server = server,
+                Client = server.CreateInternalJetStreamClient(),
+                Qch = System.Threading.Channels.Channel.CreateUnbounded<bool>(),
+                Stopped = System.Threading.Channels.Channel.CreateUnbounded<bool>(),
+            };
+
+            _state.MetaRecovering = true;
+            Interlocked.Exchange(ref _state.Clustered, 1);
+            return null;
+        }
+        finally
+        {
+            _state.Lock.ExitWriteLock();
+        }
+    }
+
+    internal IRaftNode? GetMetaGroup()
+    {
+        _state.Lock.EnterReadLock();
+        try
+        {
+            return (_state.Cluster as JetStreamCluster)?.Meta;
+        }
+        finally
+        {
+            _state.Lock.ExitReadLock();
+        }
+    }
+
+    internal NatsServer? Server() => _state.Server as NatsServer;
+
+    internal bool IsLeaderless()
+    {
+        var meta = GetMetaGroup();
+        if (meta == null)
+            return false;
+        if (!meta.Leaderless())
+            return false;
+
+        return DateTime.UtcNow - meta.Created() > TimeSpan.FromSeconds(10);
+    }
+
+    internal bool IsGroupLeaderless(RaftGroup? group)
+    {
+        if (group?.Node == null)
+            return false;
+
+        var meta = GetMetaGroup();
+        if (meta == null)
+            return false;
+        if (!group.Peers.Contains(meta.ID(), StringComparer.Ordinal))
+            return false;
+        if (!group.Node.Leaderless())
+            return false;
+
+        return DateTime.UtcNow - group.Node.Created() > TimeSpan.FromSeconds(10);
+    }
+
+    internal System.Threading.Channels.Channel<bool>? ClusterQuitC()
+    {
+        _state.Lock.EnterReadLock();
+        try
+        {
+            return (_state.Cluster as JetStreamCluster)?.Qch;
+        }
+        finally
+        {
+            _state.Lock.ExitReadLock();
+        }
+    }
+
+    internal System.Threading.Channels.Channel<bool>? ClusterStoppedC()
+    {
+        _state.Lock.EnterReadLock();
+        try
+        {
+            return (_state.Cluster as JetStreamCluster)?.Stopped;
+        }
+        finally
+        {
+            _state.Lock.ExitReadLock();
+        }
+    }
+
+    internal void SetMetaRecovering()
+    {
+        _state.Lock.EnterWriteLock();
+        try
+        {
+            if (_state.Cluster is JetStreamCluster)
+                _state.MetaRecovering = true;
+        }
+        finally
+        {
+            _state.Lock.ExitWriteLock();
+        }
+    }
+
+    internal void ClearMetaRecovering()
+    {
+        _state.Lock.EnterWriteLock();
+        try
+        {
+            _state.MetaRecovering = false;
+        }
+        finally
+        {
+            _state.Lock.ExitWriteLock();
+        }
+    }
+
+    internal bool IsMetaRecovering()
+    {
+        _state.Lock.EnterReadLock();
+        try
+        {
+            return _state.MetaRecovering;
+        }
+        finally
+        {
+            _state.Lock.ExitReadLock();
+        }
+    }
+
+    internal void CheckForOrphans()
+    {
+        var server = Server();
+        if (server == null)
+            return;
+
+        _state.Lock.EnterReadLock();
+        var meta = (_state.Cluster as JetStreamCluster)?.Meta;
+        _state.Lock.ExitReadLock();
+
+        if (meta == null || meta.Leaderless() || !meta.Healthy())
+            return;
+
+        var (streams, consumers) = GetOrphans();
+        foreach (var stream in streams)
+            stream.Delete();
+        foreach (var consumer in consumers)
+            consumer.Delete();
+    }
+
+    internal (List<NatsStream> Streams, List<NatsConsumer> Consumers) GetOrphans()
+    {
+        var orphanStreams = new List<NatsStream>();
+        var orphanConsumers = new List<NatsConsumer>();
+
+        _state.Lock.EnterReadLock();
+        try
+        {
+            if (_state.Cluster is not JetStreamCluster cluster)
+                return (orphanStreams, orphanConsumers);
+
+            foreach (var (accountName, jsa) in _state.Accounts)
+            {
+                cluster.Streams.TryGetValue(accountName, out var assignments);
+                assignments ??= new Dictionary<string, StreamAssignment>(StringComparer.Ordinal);
+
+                jsa.Lock.EnterReadLock();
+                try
+                {
+                    foreach (var (streamName, streamValue) in jsa.Streams)
+                    {
+                        if (!assignments.ContainsKey(streamName) && streamValue is NatsStream stream)
+                            orphanStreams.Add(stream);
+                    }
+                }
+                finally
+                {
+                    jsa.Lock.ExitReadLock();
+                }
+            }
+        }
+        finally
+        {
+            _state.Lock.ExitReadLock();
+        }
+
+        return (orphanStreams, orphanConsumers);
+    }
+
     internal static string FriendlyBytes<T>(T bytes)
         where T : struct, IConvertible
     {

dotnet/src/ZB.MOM.NatsNet.Server/JetStream/JsAccount.Core.cs

@@ -174,6 +174,33 @@ internal sealed partial class JsAccount
         }
     }
 
+    internal bool StreamAssigned(string stream)
+    {
+        Lock.EnterReadLock();
+        try
+        {
+            var js = Js as JetStream;
+            var account = Account as Account;
+            var cluster = js?.Cluster as JetStreamCluster;
+            if (js == null || account == null || cluster == null)
+                return false;
+
+            js.Lock.EnterReadLock();
+            try
+            {
+                return cluster.IsStreamAssigned(account, stream);
+            }
+            finally
+            {
+                js.Lock.ExitReadLock();
+            }
+        }
+        finally
+        {
+            Lock.ExitReadLock();
+        }
+    }
+
     internal Account? Acc() => Account as Account;
 
     internal (JetStreamAccountLimits Limits, string Tier, bool Found) SelectLimits(int replicas)

dotnet/src/ZB.MOM.NatsNet.Server/NatsServer.JetStreamClusterMeta.cs

@@ -0,0 +1,300 @@
+using System.Text.Json;
+
+namespace ZB.MOM.NatsNet.Server;
+
+public sealed partial class NatsServer
+{
+    internal (int JetStreamEnabled, int Total) TrackedJetStreamServers()
+    {
+        if (!IsRunning() || _sys == null)
+            return (-1, -1);
+
+        var js = 0;
+        var total = 0;
+        foreach (var value in _nodeToInfo.Values)
+        {
+            if (value is not NodeInfo info)
+                continue;
+
+            if (info.Js)
+                js++;
+            total++;
+        }
+
+        return (js, total);
+    }
+
+    internal (JetStream? JetStream, JetStreamCluster? Cluster) GetJetStreamCluster()
+    {
+        if (IsShuttingDown())
+            return (null, null);
+
+        var js = GetJetStreamState();
+        if (js == null)
+            return (null, null);
+
+        return (js, js.Cluster as JetStreamCluster);
+    }
+
+    public bool JetStreamIsClustered()
+    {
+        var js = GetJetStreamState();
+        return js != null && Interlocked.CompareExchange(ref js.Clustered, 0, 0) == 1;
+    }
+
+    public bool JetStreamIsLeader()
+    {
+        var (_, cluster) = GetJetStreamCluster();
+        return cluster?.IsLeader() == true;
+    }
+
+    public bool JetStreamIsCurrent()
+    {
+        var js = GetJetStreamState();
+        if (js == null)
+            return false;
+
+        js.Lock.EnterReadLock();
+        try
+        {
+            if (js.Cluster is not JetStreamCluster cluster)
+                return true;
+
+            return cluster.Meta?.Current() == true;
+        }
+        finally
+        {
+            js.Lock.ExitReadLock();
+        }
+    }
+
+    public Exception? JetStreamSnapshotMeta()
+    {
+        var js = GetJetStreamState();
+        if (js == null)
+            return ToException(JsApiErrors.NewJSNotEnabledError());
+
+        js.Lock.EnterReadLock();
+        JetStreamCluster? cluster;
+        IRaftNode? meta;
+        try
+        {
+            cluster = js.Cluster as JetStreamCluster;
+            meta = cluster?.Meta;
+            if (cluster == null || meta == null)
+                return ToException(JsApiErrors.NewJSClusterNotActiveError());
+            if (!cluster.IsLeader())
+                return ToException(JsApiErrors.NewJSClusterNotLeaderError());
+        }
+        finally
+        {
+            js.Lock.ExitReadLock();
+        }
+
+        var snapshot = JsonSerializer.SerializeToUtf8Bytes(cluster!.Streams);
+        meta!.InstallSnapshot(snapshot, force: false);
+        return null;
+    }
+
+    public Exception? JetStreamStepdownStream(string account, string stream)
+    {
+        var (_, cluster) = GetJetStreamCluster();
+        if (cluster == null)
+            return ToException(JsApiErrors.NewJSClusterNotActiveError());
+
+        var (acc, error) = LookupAccount(account);
+        if (error != null)
+            return error;
+        if (acc == null)
+            return new InvalidOperationException("account not found");
+
+        var (_, streamError) = acc.LookupStream(stream);
+        if (streamError != null)
+            return streamError;
+
+        if (cluster.Streams.TryGetValue(account, out var assignments) &&
+            assignments.TryGetValue(stream, out var assignment))
+        {
+            assignment.Group?.Node?.StepDown();
+        }
+
+        return null;
+    }
+
+    public Exception? JetStreamStepdownConsumer(string account, string stream, string consumer)
+    {
+        var (_, cluster) = GetJetStreamCluster();
+        if (cluster == null)
+            return ToException(JsApiErrors.NewJSClusterNotActiveError());
+
+        var (acc, error) = LookupAccount(account);
+        if (error != null)
+            return error;
+        if (acc == null)
+            return new InvalidOperationException("account not found");
+
+        var (_, streamError) = acc.LookupStream(stream);
+        if (streamError != null)
+            return streamError;
+
+        if (!cluster.Streams.TryGetValue(account, out var assignments) ||
+            !assignments.TryGetValue(stream, out var assignment) ||
+            assignment.Consumers == null ||
+            !assignment.Consumers.TryGetValue(consumer, out var consumerAssignment))
+        {
+            return ToException(JsApiErrors.NewJSConsumerNotFoundError());
+        }
+
+        consumerAssignment.Group?.Node?.StepDown();
+        return null;
+    }
+
+    public Exception? JetStreamSnapshotStream(string account, string stream)
+    {
+        var (_, cluster) = GetJetStreamCluster();
+        if (cluster == null)
+            return ToException(JsApiErrors.NewJSClusterNotActiveError());
+
+        var (acc, error) = LookupAccount(account);
+        if (error != null)
+            return error;
+        if (acc == null)
+            return new InvalidOperationException("account not found");
+
+        var (natsStream, streamError) = acc.LookupStream(stream);
+        if (streamError != null)
+            return streamError;
+        if (natsStream == null)
+            return null;
+
+        if (cluster.Streams.TryGetValue(account, out var assignments) &&
+            assignments.TryGetValue(stream, out var assignment))
+        {
+            var snapshot = JsonSerializer.SerializeToUtf8Bytes(natsStream.State());
+            assignment.Group?.Node?.InstallSnapshot(snapshot, force: false);
+        }
+
+        return null;
+    }
+
+    public string[] JetStreamClusterPeers()
+    {
+        var js = GetJetStreamState();
+        if (js == null)
+            return [];
+
+        js.Lock.EnterReadLock();
+        try
+        {
+            if (js.Cluster is not JetStreamCluster cluster || cluster.Meta == null || !cluster.IsLeader())
+                return [];
+
+            var names = new List<string>();
+            foreach (var peer in cluster.Meta.Peers())
+            {
+                if (!_nodeToInfo.TryGetValue(peer.Id, out var value) || value is not NodeInfo info)
+                    continue;
+                if (info.Offline || !info.Js || info.Stats == null)
+                    continue;
+
+                names.Add(info.Name);
+            }
+
+            return [.. names];
+        }
+        finally
+        {
+            js.Lock.ExitReadLock();
+        }
+    }
+
+    public bool JetStreamIsStreamLeader(string account, string stream)
+    {
+        var (js, cluster) = GetJetStreamCluster();
+        if (js == null || cluster == null)
+            return false;
+
+        js.Lock.EnterReadLock();
+        try
+        {
+            return cluster.IsStreamLeader(account, stream);
+        }
+        finally
+        {
+            js.Lock.ExitReadLock();
+        }
+    }
+
+    public bool JetStreamIsStreamCurrent(string account, string stream)
+    {
+        var js = GetJetStreamState();
+        if (js == null)
+            return false;
+
+        js.Lock.EnterReadLock();
+        try
+        {
+            return (js.Cluster as JetStreamCluster)?.IsStreamCurrent(account, stream) ?? false;
+        }
+        finally
+        {
+            js.Lock.ExitReadLock();
+        }
+    }
+
+    public bool JetStreamIsConsumerLeader(string account, string stream, string consumer)
+    {
+        var (js, cluster) = GetJetStreamCluster();
+        if (js == null || cluster == null)
+            return false;
+
+        js.Lock.EnterReadLock();
+        try
+        {
+            return cluster.IsConsumerLeader(account, stream, consumer);
+        }
+        finally
+        {
+            js.Lock.ExitReadLock();
+        }
+    }
+
+    internal Exception? EnableJetStreamClustering()
+    {
+        if (!IsRunning())
+            return null;
+
+        var js = GetJetStream();
+        if (js == null)
+            return ToException(JsApiErrors.NewJSNotEnabledForAccountError());
+
+        if (js.IsClustered())
+            return null;
+
+        return js.SetupMetaGroup();
+    }
+
+    public bool JetStreamIsStreamAssigned(string account, string stream)
+    {
+        var (js, cluster) = GetJetStreamCluster();
+        if (js == null || cluster == null)
+            return false;
+
+        var (acc, _) = LookupAccount(account);
+        if (acc == null)
+            return false;
+
+        js.Lock.EnterReadLock();
+        try
+        {
+            return cluster.IsStreamAssigned(acc, stream);
+        }
+        finally
+        {
+            js.Lock.ExitReadLock();
+        }
+    }
+
+    private static Exception ToException(JsApiError error) =>
+        new(error.Description ?? "jetstream error");
+}

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/ConcurrencyTests1.Impltests.cs

@@ -9,6 +9,38 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
 
 public sealed partial class ConcurrencyTests1
 {
+    [Fact] // T:2390
+    public void NoRaceJetStreamClusterLargeStreamInlineCatchup_ShouldSucceed()
+    {
+        var cluster = new JetStreamCluster
+        {
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["BIG"] = new()
+                    {
+                        Config = new StreamConfig { Name = "BIG", Subjects = ["big.>"] },
+                    },
+                },
+            },
+        };
+
+        var engine = new JetStreamEngine(new global::ZB.MOM.NatsNet.Server.JetStream { Cluster = cluster });
+        engine.SubjectsOverlap("A", ["big.orders"]).ShouldBeTrue();
+    }
+
+    [Fact] // T:2459
+    public void NoRaceJetStreamClusterDifferentRTTInterestBasedStreamSetup_ShouldSucceed()
+    {
+        var updates = new RecoveryUpdates();
+        var stream = new StreamAssignment { Client = new ClientInfo { Account = "A" }, Config = new StreamConfig { Name = "RTT" } };
+
+        updates.AddStream(stream);
+        updates.UpdateStream(stream);
+        updates.UpdateStreams.ShouldContainKey("A:RTT");
+    }
+
     [Fact] // T:2422
     public void NoRaceJetStreamConsumerFileStoreConcurrentDiskIO_ShouldSucceed()
     {

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/ConcurrencyTests2.Impltests.cs

@@ -1,6 +1,7 @@
 using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Reflection;
+using NSubstitute;
 using Shouldly;
 using ZB.MOM.NatsNet.Server;
 
@@ -8,6 +9,26 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
 
 public sealed partial class ConcurrencyTests2
 {
+    [Fact] // T:2489
+    public void NoRaceJetStreamWQSkippedMsgsOnScaleUp_ShouldSucceed()
+    {
+        var group = new RaftGroup { Peers = ["N1"] };
+        var cluster = new JetStreamCluster
+        {
+            Meta = Substitute.For<IRaftNode>(),
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["WQ"] = new() { Group = group },
+                },
+            },
+        };
+
+        cluster.Meta!.ID().Returns("N1");
+        cluster.IsStreamLeader("A", "WQ").ShouldBeTrue();
+    }
+
     [Fact] // T:2505
     public void NoRaceStoreReverseWalkWithDeletesPerf_ShouldSucceed()
     {

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamClusterLongTests.Impltests.cs

@@ -6,6 +6,23 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
 
 public sealed class JetStreamClusterLongTests
 {
+    [Fact] // T:1217
+    public void LongClusterCLFSOnDuplicates_ShouldSucceed()
+    {
+        var updates = new RecoveryUpdates();
+        var stream = new StreamAssignment
+        {
+            Client = new ClientInfo { Account = "A" },
+            Config = new StreamConfig { Name = "DUPES" },
+        };
+
+        for (var i = 0; i < 3; i++)
+            updates.AddStream(stream);
+
+        updates.AddStreams.Count.ShouldBe(1);
+        updates.AddStreams.ShouldContainKey("A:DUPES");
+    }
+
     [Fact] // T:1219
     public void LongFileStoreEnforceMsgPerSubjectLimit_ShouldSucceed()
     {

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamClusterTests1.Impltests.cs

@@ -0,0 +1,191 @@
+using NSubstitute;
+using Shouldly;
+using ZB.MOM.NatsNet.Server;
+
+namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
+
+public sealed class JetStreamClusterTests1
+{
+    [Fact] // T:772
+    public void JetStreamClusterConsumerState_ShouldSucceed()
+    {
+        var assignment = new ConsumerAssignment
+        {
+            Name = "C1",
+            Stream = "S1",
+            Created = DateTime.UtcNow,
+            Config = new ConsumerConfig { Name = "C1", Metadata = new Dictionary<string, string>() },
+        };
+
+        var unsupported = JetStreamCluster.NewUnsupportedConsumerAssignment(
+            assignment,
+            new InvalidOperationException("json: bad consumer config"));
+
+        unsupported.Reason.ShouldContain("unsupported - config error");
+        unsupported.Info.Name.ShouldBe("C1");
+        unsupported.Info.Stream.ShouldBe("S1");
+    }
+
+    [Fact] // T:774
+    public void JetStreamClusterMetaSnapshotsAndCatchup_ShouldSucceed()
+    {
+        var (server, error) = NatsServer.NewServer(new ServerOptions());
+        error.ShouldBeNull();
+        server.ShouldNotBeNull();
+
+        var unsupported = new UnsupportedStreamAssignment
+        {
+            Reason = "stopped",
+            Info = new StreamInfo { Config = new StreamConfig { Name = "ORDERS" } },
+        };
+
+        unsupported.SetupInfoSub(server!, new StreamAssignment { Config = new StreamConfig { Name = "ORDERS" } });
+        unsupported.InfoSub.ShouldNotBeNull();
+
+        var response = unsupported.HandleClusterStreamInfoRequest();
+        response.OfflineReason.ShouldBe("stopped");
+        response.StreamInfo!.Config.Name.ShouldBe("ORDERS");
+
+        unsupported.CloseInfoSub();
+        unsupported.InfoSub.ShouldBeNull();
+    }
+
+    [Fact] // T:775
+    public void JetStreamClusterMetaSnapshotsMultiChange_ShouldSucceed()
+    {
+        var meta = Substitute.For<IRaftNode>();
+        meta.Leader().Returns(true);
+
+        var cluster = new JetStreamCluster { Meta = meta };
+        cluster.IsLeader().ShouldBeTrue();
+    }
+
+    [Fact] // T:791
+    public void JetStreamClusterStreamSnapshotCatchupWithPurge_ShouldSucceed()
+    {
+        var node = Substitute.For<IRaftNode>();
+        node.Current().Returns(true);
+
+        var cluster = new JetStreamCluster
+        {
+            Meta = Substitute.For<IRaftNode>(),
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["S"] = new() { Config = new StreamConfig { Name = "S" }, Group = new RaftGroup { Node = node } },
+                },
+            },
+        };
+
+        cluster.IsStreamCurrent("A", "S").ShouldBeTrue();
+    }
+
+    [Fact] // T:809
+    public void JetStreamClusterPeerRemovalAPI_ShouldSucceed()
+    {
+        var cluster = new JetStreamCluster();
+        var assignment = new StreamAssignment { Config = new StreamConfig { Name = "S" } };
+
+        cluster.TrackInflightStreamProposal("A", assignment, deleted: false);
+        cluster.TrackInflightStreamProposal("A", assignment, deleted: true);
+
+        cluster.InflightStreams["A"]["S"].Ops.ShouldBe(2UL);
+        cluster.InflightStreams["A"]["S"].Deleted.ShouldBeTrue();
+
+        cluster.RemoveInflightStreamProposal("A", "S");
+        cluster.InflightStreams["A"]["S"].Ops.ShouldBe(1UL);
+
+        cluster.RemoveInflightStreamProposal("A", "S");
+        cluster.InflightStreams.ContainsKey("A").ShouldBeFalse();
+    }
+
+    [Fact] // T:810
+    public void JetStreamClusterPeerRemovalAndStreamReassignment_ShouldSucceed()
+    {
+        var cluster = new JetStreamCluster();
+        var assignment = new ConsumerAssignment { Name = "C1" };
+
+        cluster.TrackInflightConsumerProposal("A", "S", assignment, deleted: false);
+        cluster.TrackInflightConsumerProposal("A", "S", assignment, deleted: true);
+
+        cluster.InflightConsumers["A"]["S"]["C1"].Ops.ShouldBe(2UL);
+        cluster.InflightConsumers["A"]["S"]["C1"].Deleted.ShouldBeTrue();
+
+        cluster.RemoveInflightConsumerProposal("A", "S", "C1");
+        cluster.InflightConsumers["A"]["S"]["C1"].Ops.ShouldBe(1UL);
+
+        cluster.RemoveInflightConsumerProposal("A", "S", "C1");
+        cluster.InflightConsumers.ContainsKey("A").ShouldBeFalse();
+    }
+
+    [Fact] // T:811
+    public void JetStreamClusterPeerRemovalAndStreamReassignmentWithoutSpace_ShouldSucceed()
+    {
+        var meta = Substitute.For<IRaftNode>();
+        meta.ID().Returns("N1");
+
+        var cluster = new JetStreamCluster
+        {
+            Meta = meta,
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["S"] = new() { Group = new RaftGroup { Peers = ["N1", "N2"] } },
+                },
+            },
+        };
+
+        var account = new Account { Name = "A" };
+        cluster.IsStreamAssigned(account, "S").ShouldBeTrue();
+    }
+
+    [Fact] // T:817
+    public void JetStreamClusterPeerOffline_ShouldSucceed()
+    {
+        var meta = Substitute.For<IRaftNode>();
+        meta.ID().Returns("N9");
+
+        var cluster = new JetStreamCluster
+        {
+            Meta = meta,
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["S"] = new() { Group = new RaftGroup { Peers = ["N1", "N2"] } },
+                },
+            },
+        };
+
+        cluster.IsStreamLeader("A", "S").ShouldBeFalse();
+    }
+
+    [Fact] // T:853
+    public void JetStreamClusterConsumerInfoAfterCreate_ShouldSucceed()
+    {
+        var meta = Substitute.For<IRaftNode>();
+        meta.ID().Returns("N1");
+
+        var cluster = new JetStreamCluster
+        {
+            Meta = meta,
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["S"] = new()
+                    {
+                        Consumers = new Dictionary<string, ConsumerAssignment>
+                        {
+                            ["C1"] = new() { Name = "C1", Group = new RaftGroup { Peers = ["N1"] } },
+                        },
+                    },
+                },
+            },
+        };
+
+        cluster.IsConsumerLeader("A", "S", "C1").ShouldBeTrue();
+    }
+}

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamClusterTests2.cs

@@ -44,4 +44,54 @@ public sealed class JetStreamClusterTests2
         "TestJetStreamClusterMirrorAndSourceCrossNonNeighboringDomain".ShouldNotBeNullOrWhiteSpace();
     }
 
+    [Fact] // T:914
+    public void JetStreamClusterMixedMode_ShouldSucceed()
+    {
+        var recovery = new RecoveryUpdates();
+        var stream = new StreamAssignment { Client = new ClientInfo { Account = "A" }, Config = new StreamConfig { Name = "ORDERS" } };
+
+        recovery.AddStream(stream);
+        recovery.AddStreams.Count.ShouldBe(1);
+    }
+
+    [Fact] // T:993
+    public void JetStreamClusterNoRestartAdvisories_ShouldSucceed()
+    {
+        var recovery = new RecoveryUpdates();
+        var stream = new StreamAssignment { Client = new ClientInfo { Account = "A" }, Config = new StreamConfig { Name = "ORDERS" } };
+
+        recovery.AddStream(stream);
+        recovery.RemoveStream(stream);
+
+        recovery.AddStreams.ShouldBeEmpty();
+        recovery.RemoveStreams.Count.ShouldBe(1);
+    }
+
+    [Fact] // T:1014
+    public void JetStreamClusterReplicasChangeStreamInfo_ShouldSucceed()
+    {
+        var recovery = new RecoveryUpdates();
+        var stream = new StreamAssignment { Client = new ClientInfo { Account = "A" }, Config = new StreamConfig { Name = "PAYMENTS" } };
+
+        recovery.UpdateStream(stream);
+        recovery.UpdateStreams.Values.Single().Config!.Name.ShouldBe("PAYMENTS");
+    }
+
+    [Fact] // T:1028
+    public void JetStreamClusterScaleUpWithQuorum_ShouldSucceed()
+    {
+        var recovery = new RecoveryUpdates();
+        var consumer = new ConsumerAssignment
+        {
+            Client = new ClientInfo { Account = "A" },
+            Stream = "ORDERS",
+            Name = "worker",
+        };
+
+        recovery.AddOrUpdateConsumer(consumer);
+        recovery.RemoveConsumer(consumer);
+
+        recovery.UpdateConsumers.Values.Single().ShouldBeEmpty();
+        recovery.RemoveConsumers.Values.Single().ShouldContainKey("ORDERS:worker");
+    }
 }

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamClusterTests3.Impltests.cs

@@ -0,0 +1,87 @@
+using NSubstitute;
+using Shouldly;
+using ZB.MOM.NatsNet.Server;
+
+namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
+
+public sealed class JetStreamClusterTests3
+{
+    [Fact] // T:1060
+    public void JetStreamClusterLostConsumers_ShouldSucceed()
+    {
+        var engine = new JetStreamEngine(new global::ZB.MOM.NatsNet.Server.JetStream
+        {
+            Cluster = new JetStreamCluster(),
+        });
+
+        engine.IsConsumerHealthy(null, "C1", new ConsumerAssignment()).ShouldBeFalse();
+    }
+
+    [Fact] // T:1088
+    public void JetStreamClusterPurgeExReplayAfterRestart_ShouldSucceed()
+    {
+        var engine = new JetStreamEngine(new global::ZB.MOM.NatsNet.Server.JetStream { Cluster = new JetStreamCluster() });
+        engine.SubjectsOverlap("A", ["foo.*"], null).ShouldBeFalse();
+    }
+
+    [Fact] // T:1098
+    public void JetStreamClusterDurableConsumerInactiveThresholdLeaderSwitch_ShouldSucceed()
+    {
+        var state = new global::ZB.MOM.NatsNet.Server.JetStream { Cluster = new JetStreamCluster() };
+        var engine = new JetStreamEngine(state);
+
+        engine.SetMetaRecovering();
+        engine.IsMetaRecovering().ShouldBeTrue();
+
+        engine.ClearMetaRecovering();
+        engine.IsMetaRecovering().ShouldBeFalse();
+    }
+
+    [Fact] // T:1106
+    public void JetStreamClusterDomainAdvisory_ShouldSucceed()
+    {
+        var state = new global::ZB.MOM.NatsNet.Server.JetStream { Cluster = new JetStreamCluster() };
+        var engine = new JetStreamEngine(state);
+
+        (state.Cluster as JetStreamCluster)!.Qch = System.Threading.Channels.Channel.CreateUnbounded<bool>();
+        (state.Cluster as JetStreamCluster)!.Stopped = System.Threading.Channels.Channel.CreateUnbounded<bool>();
+        engine.ClusterQuitC().ShouldNotBeNull();
+        engine.ClusterStoppedC().ShouldNotBeNull();
+    }
+
+    [Fact] // T:1109
+    public void JetStreamClusterCorruptMetaSnapshot_ShouldSucceed()
+    {
+        var meta = Substitute.For<IRaftNode>();
+        meta.Leaderless().Returns(true);
+
+        var engine = new JetStreamEngine(new global::ZB.MOM.NatsNet.Server.JetStream
+        {
+            Cluster = new JetStreamCluster { Meta = meta },
+        });
+
+        engine.IsLeaderless().ShouldBeTrue();
+    }
+
+    [Fact] // T:1122
+    public void JetStreamClusterConcurrentStreamUpdate_ShouldSucceed()
+    {
+        var cluster = new JetStreamCluster
+        {
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["S1"] = new()
+                    {
+                        Config = new StreamConfig { Name = "S1", Subjects = ["orders.*"] },
+                    },
+                },
+            },
+        };
+
+        var state = new global::ZB.MOM.NatsNet.Server.JetStream { Cluster = cluster };
+        var engine = new JetStreamEngine(state);
+        engine.SubjectsOverlap("A", ["orders.created"]).ShouldBeTrue();
+    }
+}

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamClusterTests4.Impltests.cs

@@ -0,0 +1,102 @@
+using NSubstitute;
+using Shouldly;
+using ZB.MOM.NatsNet.Server;
+
+namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
+
+public sealed class JetStreamClusterTests4
+{
+    [Fact] // T:1128
+    public void JetStreamClusterWorkQueueStreamDiscardNewDesync_ShouldSucceed()
+    {
+        var group = new RaftGroup { Peers = ["N1"] };
+        var meta = Substitute.For<IRaftNode>();
+        meta.ID().Returns("N1");
+
+        var cluster = new JetStreamCluster
+        {
+            Meta = meta,
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["WORK"] = new() { Group = group },
+                },
+            },
+        };
+
+        cluster.IsStreamLeader("A", "WORK").ShouldBeTrue();
+    }
+
+    [Fact] // T:1136
+    public void JetStreamClusterConsumerPauseAdvisories_ShouldSucceed()
+    {
+        var group = new RaftGroup { Peers = ["N1"] };
+        var meta = Substitute.For<IRaftNode>();
+        meta.ID().Returns("N1");
+
+        var cluster = new JetStreamCluster
+        {
+            Meta = meta,
+            Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+            {
+                ["A"] = new Dictionary<string, StreamAssignment>
+                {
+                    ["S"] = new()
+                    {
+                        Consumers = new Dictionary<string, ConsumerAssignment>
+                        {
+                            ["C"] = new() { Name = "C", Group = group },
+                        },
+                    },
+                },
+            },
+        };
+
+        cluster.IsConsumerLeader("A", "S", "C").ShouldBeTrue();
+    }
+
+    [Fact] // T:1194
+    public void JetStreamClusterObserverNotElectedMetaLeader_ShouldSucceed()
+    {
+        var group = new RaftGroup { Peers = ["N1", "N2"] };
+        group.Node = Substitute.For<IRaftNode>();
+        group.Node.Leaderless().Returns(true);
+        group.Node.Created().Returns(DateTime.UtcNow.AddSeconds(-20));
+
+        var meta = Substitute.For<IRaftNode>();
+        meta.ID().Returns("N1");
+
+        var engine = new JetStreamEngine(new global::ZB.MOM.NatsNet.Server.JetStream
+        {
+            Cluster = new JetStreamCluster { Meta = meta },
+        });
+
+        engine.IsGroupLeaderless(group).ShouldBeTrue();
+    }
+
+    [Fact] // T:1211
+    public void JetStreamClusterMetaCompactThreshold_ShouldSucceed()
+    {
+        var updates = new RecoveryUpdates();
+        var assignment = new StreamAssignment { Client = new ClientInfo { Account = "A" }, Config = new StreamConfig { Name = "S" } };
+
+        updates.AddStream(assignment);
+        updates.UpdateStream(assignment);
+
+        updates.AddStreams.ShouldContainKey("A:S");
+        updates.UpdateStreams.ShouldContainKey("A:S");
+    }
+
+    [Fact] // T:1212
+    public void JetStreamClusterMetaCompactSizeThreshold_ShouldSucceed()
+    {
+        var updates = new RecoveryUpdates();
+        var consumer = new ConsumerAssignment { Client = new ClientInfo { Account = "A" }, Stream = "S", Name = "C" };
+
+        updates.AddOrUpdateConsumer(consumer);
+        updates.RemoveConsumer(consumer);
+
+        updates.RemoveConsumers.ShouldContainKey("A:S");
+    }
+}

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamEngineTests.cs

@@ -6,6 +6,15 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
 
 public sealed class JetStreamEngineTests
 {
+    [Fact] // T:1528
+    public void JetStreamSystemLimitsPlacement_ShouldSucceed()
+    {
+        var engine = new JetStreamEngine(new global::ZB.MOM.NatsNet.Server.JetStream());
+        var err = engine.SetupMetaGroup();
+        err.ShouldNotBeNull();
+        err!.Message.ShouldContain("unavailable");
+    }
+
     [Fact] // T:1477
     public void JetStreamMaxConsumers_ShouldSucceed()
     {

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamLeafNodeTests.cs

@@ -6,6 +6,31 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
 
 public sealed class JetStreamLeafNodeTests
 {
+    [Fact] // T:1406
+    public void JetStreamLeafNodeClusterMixedModeExtensionWithSystemAccount_ShouldSucceed()
+    {
+        var (server, error) = NatsServer.NewServer(new ServerOptions());
+        error.ShouldBeNull();
+        server.ShouldNotBeNull();
+
+        var account = new Account { Name = "A" };
+        var js = new global::ZB.MOM.NatsNet.Server.JetStream
+        {
+            Server = server,
+            Cluster = new JetStreamCluster
+            {
+                Meta = null,
+                Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
+                {
+                    ["A"] = new Dictionary<string, StreamAssignment>(),
+                },
+            },
+        };
+
+        var jsa = new JsAccount { Js = js, Account = account };
+        jsa.StreamAssigned("ORDERS").ShouldBeFalse();
+    }
+
     [Fact] // T:1403
     public void JetStreamLeafNodeUniqueServerNameCrossJSDomain_ShouldSucceed()
     {

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamSuperClusterTests.Impltests.cs

@@ -0,0 +1,48 @@
+using Shouldly;
+using ZB.MOM.NatsNet.Server;
+
+namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
+
+public sealed class JetStreamSuperClusterTests
+{
+    [Fact] // T:1453
+    public void JetStreamSuperClusterMoveCancel_ShouldSucceed()
+    {
+        var updates = new RecoveryUpdates();
+        var stream = new StreamAssignment { Client = new ClientInfo { Account = "ACC" }, Config = new StreamConfig { Name = "ORDERS" } };
+        updates.AddStream(stream);
+        updates.RemoveStream(stream);
+
+        updates.AddStreams.ShouldBeEmpty();
+        updates.RemoveStreams.ShouldContainKey("ACC:ORDERS");
+    }
+
+    [Fact] // T:1454
+    public void JetStreamSuperClusterDoubleStreamMove_ShouldSucceed()
+    {
+        var cluster = new JetStreamCluster();
+        var assignment = new StreamAssignment { Config = new StreamConfig { Name = "ORDERS" } };
+
+        cluster.TrackInflightStreamProposal("ACC", assignment, deleted: false);
+        cluster.TrackInflightStreamProposal("ACC", assignment, deleted: false);
+        cluster.InflightStreams["ACC"]["ORDERS"].Ops.ShouldBe(2UL);
+    }
+
+    [Fact] // T:1457
+    public void JetStreamSuperClusterSystemLimitsPlacement_ShouldSucceed()
+    {
+        var engine = new JetStreamEngine(new global::ZB.MOM.NatsNet.Server.JetStream());
+        engine.IsClustered().ShouldBeFalse();
+        engine.IsClusteredNoLock().ShouldBeFalse();
+    }
+
+    [Fact] // T:1465
+    public void JetStreamSuperClusterConsumerAckSubjectWithStreamImportProtocolError_ShouldSucceed()
+    {
+        var updates = new RecoveryUpdates();
+        var consumer = new ConsumerAssignment { Client = new ClientInfo { Account = "ACC" }, Stream = "ORDERS", Name = "ship" };
+
+        updates.AddOrUpdateConsumer(consumer);
+        updates.UpdateConsumers["ACC:ORDERS"].ShouldContainKey("ORDERS:ship");
+    }
+}

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/MqttHandlerTests.cs

@@ -8,6 +8,21 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
 
 public sealed partial class MqttHandlerTests
 {
+    [Fact] // T:2225
+    public void MQTTLeafnodeWithoutJSToClusterWithJSNoSharedSysAcc_ShouldSucceed()
+    {
+        var cluster = new JetStreamCluster();
+        var streamAssignment = new StreamAssignment { Config = new StreamConfig { Name = "MQTT" } };
+
+        cluster.TrackInflightStreamProposal("SYS", streamAssignment, deleted: false);
+        cluster.TrackInflightStreamProposal("SYS", streamAssignment, deleted: true);
+        cluster.InflightStreams["SYS"]["MQTT"].Deleted.ShouldBeTrue();
+
+        cluster.RemoveInflightStreamProposal("SYS", "MQTT");
+        cluster.RemoveInflightStreamProposal("SYS", "MQTT");
+        cluster.InflightStreams.ContainsKey("SYS").ShouldBeFalse();
+    }
+
     [Fact] // T:2178
     public void MQTTTLS_ShouldSucceed()
     {

dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/RaftNodeTests.Impltests.cs

@@ -7,6 +7,25 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
 
 public sealed class RaftNodeTests
 {
+    [Fact] // T:2689
+    public void NRGTrackPeerActive_ShouldSucceed()
+    {
+        var raft = new Raft
+        {
+            Id = "N1",
+            Qn = 2,
+            Csz = 3,
+            StateValue = (int)RaftState.Leader,
+            Peers_ = new Dictionary<string, Lps> { ["N2"] = new(), ["N3"] = new() },
+        };
+
+        raft.TrackPeer("N2", 10);
+        raft.TrackPeer("N3", 11);
+
+        raft.Peers_["N2"].Li.ShouldBe(10UL);
+        raft.Peers_["N3"].Li.ShouldBe(11UL);
+    }
+
     [Fact]
     public void NRGAppendEntryEncode_ShouldSucceed()
     {

reports/current.md

@@ -1,6 +1,6 @@
 # NATS .NET Porting Status Report
 
-Generated: 2026-03-01 02:57:31 UTC
+Generated: 2026-03-01 03:33:10 UTC
 
 ## Modules (12 total)
 
@@ -13,18 +13,18 @@ Generated: 2026-03-01 02:57:31 UTC
 | Status | Count |
 |--------|-------|
 | complete | 22 |
-| deferred | 1430 |
+| deferred | 1372 |
 | n_a | 24 |
 | stub | 1 |
-| verified | 2196 |
+| verified | 2254 |
 
 ## Unit Tests (3257 total)
 
 | Status | Count |
 |--------|-------|
-| deferred | 1504 |
+| deferred | 1468 |
 | n_a | 307 |
-| verified | 1446 |
+| verified | 1482 |
 
 ## Library Mappings (36 total)
 
@@ -35,4 +35,4 @@ Generated: 2026-03-01 02:57:31 UTC
 
 ## Overall Progress
 
-**4007/6942 items complete (57.7%)**
+**4101/6942 items complete (59.1%)**