19188141845

Committed 07 Nov 2025 01:16PM UTC coverage: 84.586% (-1.4%) from 86.033%

Build # 19188141845

Build Type

push

github

Committed by

web-flow

Commit Message

Add meta snapshot metrics to jsz monitoring (#7524)

Exposes snapshot related metrics under /jsz

```js
"meta_cluster": {
    "pending": 0,
    "snapshot": {
        "pending_entries": 1,
        "pending_size": 1314,
        "last_time": "2025-11-06T18:14:40.659678019Z", # UTC
        "last_duration": 161096363
     }
}
```

Signed-off-by: Waldemar Quevedo <wally@nats.io>

Coverage Stats

73649 of 87070 relevant lines covered (84.59%)

340562.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.3

/server/server.go

// Copyright 2012-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package server

import (
        "bytes"
        "context"
        "crypto/fips140"
        "crypto/tls"
        "encoding/json"
        "errors"
        "flag"
        "fmt"
        "io"
        "log"
        "math/rand"
        "net"
        "net/http"
        "net/url"
        "os"
        "path"
        "path/filepath"
        "reflect"
        "regexp"
        "runtime"
        "runtime/pprof"
        "strconv"
        "strings"
        "sync"
        "sync/atomic"
        "time"

        // Allow dynamic profiling.
        _ "net/http/pprof"

        "expvar"

        "github.com/klauspost/compress/s2"
        "github.com/nats-io/jwt/v2"
        "github.com/nats-io/nats-server/v2/logger"
        "github.com/nats-io/nkeys"
        "github.com/nats-io/nuid"
)

const (
        // Interval for the first PING for non client connections.
        firstPingInterval = time.Second

        // This is for the first ping for client connections.
        firstClientPingInterval = 2 * time.Second
)

// These are protocol versions sent between server connections: ROUTER, LEAF and
// GATEWAY. We may have protocol versions that have a meaning only for a certain
// type of connections, but we don't have to have separate enums for that.
// However, it is CRITICAL to not change the order of those constants since they
// are exchanged between servers. When adding a new protocol version, add to the
// end of the list, don't try to group them by connection types.
const (
        // RouteProtoZero is the original Route protocol from 2009.
        // http://nats.io/documentation/internals/nats-protocol/
        RouteProtoZero = iota
        // RouteProtoInfo signals a route can receive more then the original INFO block.
        // This can be used to update remote cluster permissions, etc...
        RouteProtoInfo
        // RouteProtoV2 is the new route/cluster protocol that provides account support.
        RouteProtoV2
        // MsgTraceProto indicates that this server understands distributed message tracing.
        MsgTraceProto
)

// Will return the latest server-to-server protocol versions, unless the
// option to override it is set.
func (s *Server) getServerProto() int {
        opts := s.getOpts()
        // Initialize with the latest protocol version.
        proto := MsgTraceProto
        // For tests, we want to be able to make this server behave
        // as an older server so check this option to see if we should override.
        if opts.overrideProto < 0 {
                // The option overrideProto is set to 0 by default (when creating an
                // Options structure). Since this is the same value than the original
                // proto RouteProtoZero, tests call setServerProtoForTest() with the
                // desired protocol level, which sets it as negative value equal to:
                // (wantedProto + 1) * -1. Here we compute back the real value.
                proto = (opts.overrideProto * -1) - 1
        }
        return proto
}

// Used by tests.
func setServerProtoForTest(wantedProto int) int {
        return (wantedProto + 1) * -1
}

// Info is the information sent to clients, routes, gateways, and leaf nodes,
// to help them understand information about this server.
type Info struct {
        ID                string   `json:"server_id"`
        Name              string   `json:"server_name"`
        Version           string   `json:"version"`
        Proto             int      `json:"proto"`
        GitCommit         string   `json:"git_commit,omitempty"`
        GoVersion         string   `json:"go"`
        Host              string   `json:"host"`
        Port              int      `json:"port"`
        Headers           bool     `json:"headers"`
        AuthRequired      bool     `json:"auth_required,omitempty"`
        TLSRequired       bool     `json:"tls_required,omitempty"`
        TLSVerify         bool     `json:"tls_verify,omitempty"`
        TLSAvailable      bool     `json:"tls_available,omitempty"`
        MaxPayload        int32    `json:"max_payload"`
        JetStream         bool     `json:"jetstream,omitempty"`
        IP                string   `json:"ip,omitempty"`
        CID               uint64   `json:"client_id,omitempty"`
        ClientIP          string   `json:"client_ip,omitempty"`
        Nonce             string   `json:"nonce,omitempty"`
        Cluster           string   `json:"cluster,omitempty"`
        Dynamic           bool     `json:"cluster_dynamic,omitempty"`
        Domain            string   `json:"domain,omitempty"`
        ClientConnectURLs []string `json:"connect_urls,omitempty"`    // Contains URLs a client can connect to.
        WSConnectURLs     []string `json:"ws_connect_urls,omitempty"` // Contains URLs a ws client can connect to.
        LameDuckMode      bool     `json:"ldm,omitempty"`
        Compression       string   `json:"compression,omitempty"`
        ConnectInfo       bool     `json:"connect_info,omitempty"`   // When true this is the server INFO response to CONNECT
        RemoteAccount     string   `json:"remote_account,omitempty"` // Lets the client or leafnode side know the remote account that they bind to.
        IsSystemAccount   bool     `json:"acc_is_sys,omitempty"`     // Indicates if the account is a system account.
        JSApiLevel        int      `json:"api_lvl,omitempty"`

        // Route Specific
        Import        *SubjectPermission `json:"import,omitempty"`
        Export        *SubjectPermission `json:"export,omitempty"`
        LNOC          bool               `json:"lnoc,omitempty"`
        LNOCU         bool               `json:"lnocu,omitempty"`
        InfoOnConnect bool               `json:"info_on_connect,omitempty"` // When true the server will respond to CONNECT with an INFO
        RoutePoolSize int                `json:"route_pool_size,omitempty"`
        RoutePoolIdx  int                `json:"route_pool_idx,omitempty"`
        RouteAccount  string             `json:"route_account,omitempty"`
        RouteAccReqID string             `json:"route_acc_add_reqid,omitempty"`
        GossipMode    byte               `json:"gossip_mode,omitempty"`

        // Gateways Specific
        Gateway           string   `json:"gateway,omitempty"`             // Name of the origin Gateway (sent by gateway's INFO)
        GatewayURLs       []string `json:"gateway_urls,omitempty"`        // Gateway URLs in the originating cluster (sent by gateway's INFO)
        GatewayURL        string   `json:"gateway_url,omitempty"`         // Gateway URL on that server (sent by route's INFO)
        GatewayCmd        byte     `json:"gateway_cmd,omitempty"`         // Command code for the receiving server to know what to do
        GatewayCmdPayload []byte   `json:"gateway_cmd_payload,omitempty"` // Command payload when needed
        GatewayNRP        bool     `json:"gateway_nrp,omitempty"`         // Uses new $GNR. prefix for mapped replies
        GatewayIOM        bool     `json:"gateway_iom,omitempty"`         // Indicate that all accounts will be switched to InterestOnly mode "right away"

        // LeafNode Specific
        LeafNodeURLs []string `json:"leafnode_urls,omitempty"` // LeafNode URLs that the server can reconnect to.

        XKey string `json:"xkey,omitempty"` // Public server's x25519 key.
}

// Server is our main struct.
type Server struct {
        // Fields accessed with atomic operations need to be 64-bit aligned
        gcid uint64
        // How often user logon fails due to the issuer account not being pinned.
        pinnedAccFail uint64
        stats
        scStats
        staleStats
        mu                  sync.RWMutex
        reloadMu            sync.RWMutex // Write-locked when a config reload is taking place ONLY
        kp                  nkeys.KeyPair
        xkp                 nkeys.KeyPair
        xpub                string
        info                Info
        configFile          string
        optsMu              sync.RWMutex
        opts                *Options
        running             atomic.Bool
        shutdown            atomic.Bool
        listener            net.Listener
        listenerErr         error
        gacc                *Account
        sys                 *internal
        sysAcc              atomic.Pointer[Account]
        js                  atomic.Pointer[jetStream]
        isMetaLeader        atomic.Bool
        jsClustered         atomic.Bool
        accounts            sync.Map
        tmpAccounts         sync.Map // Temporarily stores accounts that are being built
        activeAccounts      int32
        accResolver         AccountResolver
        clients             map[uint64]*client
        routes              map[string][]*client
        remoteRoutePoolSize map[string]int                // Map for remote's configure route pool size
        routesPoolSize      int                           // Configured pool size
        routesReject        bool                          // During reload, we may want to reject adding routes until some conditions are met
        routesNoPool        int                           // Number of routes that don't use pooling (connecting to older server for instance)
        accRoutes           map[string]map[string]*client // Key is account name, value is key=remoteID/value=route connection
        accRouteByHash      sync.Map                      // Key is account name, value is nil or a pool index
        accAddedCh          chan struct{}
        accAddedReqID       string
        leafs               map[uint64]*client
        users               map[string]*User
        nkeys               map[string]*NkeyUser
        totalClients        uint64
        closed              *closedRingBuffer
        done                chan bool
        start               time.Time
        http                net.Listener
        httpHandler         http.Handler
        httpBasePath        string
        profiler            net.Listener
        httpReqStats        map[string]uint64
        routeListener       net.Listener
        routeListenerErr    error
        routeInfo           Info
        routeResolver       netResolver
        routesToSelf        map[string]struct{}
        routeTLSName        string
        leafNodeListener    net.Listener
        leafNodeListenerErr error
        leafNodeInfo        Info
        leafNodeInfoJSON    []byte
        leafURLsMap         refCountedUrlSet
        leafNodeOpts        struct {
                resolver    netResolver
                dialTimeout time.Duration
        }
        leafRemoteCfgs     []*leafNodeCfg
        leafRemoteAccounts sync.Map
        leafNodeEnabled    bool
        leafDisableConnect bool // Used in test only
        leafNoCluster      bool // Indicate that this server has only remotes and no cluster defined

        quitCh           chan struct{}
        startupComplete  chan struct{}
        shutdownComplete chan struct{}

        // Tracking Go routines
        grMu         sync.Mutex
        grTmpClients map[uint64]*client
        grRunning    bool
        grWG         sync.WaitGroup // to wait on various go routines

        cproto     int64     // number of clients supporting async INFO
        configTime time.Time // last time config was loaded

        logging struct {
                sync.RWMutex
                logger      Logger
                trace       int32
                debug       int32
                traceSysAcc int32
        }

        clientConnectURLs []string

        // Used internally for quick look-ups.
        clientConnectURLsMap refCountedUrlSet

        // For Gateways
        gatewayListener    net.Listener // Accept listener
        gatewayListenerErr error
        gateway            *srvGateway

        // Used by tests to check that http.Servers do
        // not set any timeout.
        monitoringServer *http.Server
        profilingServer  *http.Server

        // LameDuck mode
        ldm   bool
        ldmCh chan bool

        // Trusted public operator keys.
        trustedKeys []string
        // map of trusted keys to operator setting StrictSigningKeyUsage
        strictSigningKeyUsage map[string]struct{}

        // We use this to minimize mem copies for requests to monitoring
        // endpoint /varz (when it comes from http).
        varzMu sync.Mutex
        varz   *Varz
        // This is set during a config reload if we detect that we have
        // added/removed routes. The monitoring code then check that
        // to know if it should update the cluster's URLs array.
        varzUpdateRouteURLs bool

        // Keeps a sublist of of subscriptions attached to leafnode connections
        // for the $GNR.*.*.*.> subject so that a server can send back a mapped
        // gateway reply.
        gwLeafSubs *Sublist

        // Used for expiration of mapped GW replies
        gwrm struct {
                w  int32
                ch chan time.Duration
                m  sync.Map
        }

        // For eventIDs
        eventIds *nuid.NUID

        // Websocket structure
        websocket srvWebsocket

        // MQTT structure
        mqtt srvMQTT

        // OCSP monitoring
        ocsps []*OCSPMonitor

        // OCSP peer verification (at least one TLS block)
        ocspPeerVerify bool

        // OCSP response cache
        ocsprc OCSPResponseCache

        // exporting account name the importer experienced issues with
        incompleteAccExporterMap sync.Map

        // Holds cluster name under different lock for mapping
        cnMu sync.RWMutex
        cn   string

        // For registering raft nodes with the server.
        rnMu      sync.RWMutex
        raftNodes map[string]RaftNode

        // For mapping from a raft node name back to a server name and cluster. Node has to be in the same domain.
        nodeToInfo sync.Map

        // For out of resources to not log errors too fast.
        rerrMu   sync.Mutex
        rerrLast time.Time

        connRateCounter *rateCounter

        // If there is a system account configured, to still support the $G account,
        // the server will create a fake user and add it to the list of users.
        // Keep track of what that user name is for config reload purposes.
        sysAccOnlyNoAuthUser string

        // IPQueues map
        ipQueues sync.Map

        // To limit logging frequency
        rateLimitLogging   sync.Map
        rateLimitLoggingCh chan time.Duration

        // Total outstanding catchup bytes in flight.
        gcbMu     sync.RWMutex
        gcbOut    int64
        gcbOutMax int64 // Taken from JetStreamMaxCatchup or defaultMaxTotalCatchupOutBytes
        // A global chanel to kick out stalled catchup sequences.
        gcbKick chan struct{}

        // Total outbound syncRequests
        syncOutSem chan struct{}

        // Queue to process JS API requests that come from routes (or gateways)
        jsAPIRoutedReqs *ipQueue[*jsAPIRoutedReq]

        // Delayed API responses.
        delayedAPIResponses *ipQueue[*delayedAPIResponse]

        // Whether moving NRG traffic into accounts is permitted on this server.
        // Controls whether or not the account NRG capability is set in statsz.
        // Currently used by unit tests to simulate nodes not supporting account NRG.
        accountNRGAllowed atomic.Bool

        // List of proxies trusted keys in `KeyPair` form so we can do signature
        // verification when processing incoming proxy connections.
        proxiesKeyPairs []nkeys.KeyPair
        proxiedConns    map[string]map[uint64]*client
}

// For tracking JS nodes.
type nodeInfo struct {
        name            string
        version         string
        cluster         string
        domain          string
        id              string
        tags            jwt.TagList
        cfg             *JetStreamConfig
        stats           *JetStreamStats
        offline         bool
        js              bool
        binarySnapshots bool
        accountNRG      bool
}

type stats struct {
        inMsgs           int64
        outMsgs          int64
        inBytes          int64
        outBytes         int64
        slowConsumers    int64
        staleConnections int64
        stalls           int64
}

// scStats includes the total and per connection counters of Slow Consumers.
type scStats struct {
        clients  atomic.Uint64
        routes   atomic.Uint64
        leafs    atomic.Uint64
        gateways atomic.Uint64
}

// staleStats includes the total and per connection counters of Stale Connections.
type staleStats struct {
        clients  atomic.Uint64
        routes   atomic.Uint64
        leafs    atomic.Uint64
        gateways atomic.Uint64
}

// This is used by tests so we can run all server tests with a default route
// or leafnode compression mode. For instance:
// go test -race -v ./server -cluster_compression=fast
var (
        testDefaultClusterCompression  string
        testDefaultLeafNodeCompression string
)

// Compression modes.
const (
        CompressionNotSupported   = "not supported"
        CompressionOff            = "off"
        CompressionAccept         = "accept"
        CompressionS2Auto         = "s2_auto"
        CompressionS2Uncompressed = "s2_uncompressed"
        CompressionS2Fast         = "s2_fast"
        CompressionS2Better       = "s2_better"
        CompressionS2Best         = "s2_best"
)

// defaultCompressionS2AutoRTTThresholds is the default of RTT thresholds for
// the CompressionS2Auto mode.
var defaultCompressionS2AutoRTTThresholds = []time.Duration{
        // [0..10ms] -> CompressionS2Uncompressed
        10 * time.Millisecond,
        // ]10ms..50ms] -> CompressionS2Fast
        50 * time.Millisecond,
        // ]50ms..100ms] -> CompressionS2Better
        100 * time.Millisecond,
        // ]100ms..] -> CompressionS2Best
}

// For a given user provided string, matches to one of the compression mode
// constant and updates the provided string to that constant. Returns an
// error if the provided compression mode is not known.
// The parameter `chosenModeForOn` indicates which compression mode to use
// when the user selects "on" (or enabled, true, etc..). This is because
// we may have different defaults depending on where the compression is used.
func validateAndNormalizeCompressionOption(c *CompressionOpts, chosenModeForOn string) error {
        if c == nil {
                return nil
        }
        cmtl := strings.ToLower(c.Mode)
        // First, check for the "on" case so that we set to the default compression
        // mode for that. The other switch/case will finish setup if needed (for
        // instance if the default mode is s2Auto).
        switch cmtl {
        case "on", "enabled", "true":
                cmtl = chosenModeForOn
        default:
        }
        // Check (again) with the proper mode.
        switch cmtl {
        case "not supported", "not_supported":
                c.Mode = CompressionNotSupported
        case "disabled", "off", "false":
                c.Mode = CompressionOff
        case "accept":
                c.Mode = CompressionAccept
        case "auto", "s2_auto":
                var rtts []time.Duration
                if len(c.RTTThresholds) == 0 {
                        rtts = defaultCompressionS2AutoRTTThresholds
                } else {
                        for _, n := range c.RTTThresholds {
                                // Do not error on negative, but simply set to 0
                                if n < 0 {
                                        n = 0
                                }
                                // Make sure they are properly ordered. However, it is possible
                                // to have a "0" anywhere in the list to indicate that this
                                // compression level should not be used.
                                if l := len(rtts); l > 0 && n != 0 {
                                        for _, v := range rtts {
                                                if n < v {
                                                        return fmt.Errorf("RTT threshold values %v should be in ascending order", c.RTTThresholds)
                                                }
                                        }
                                }
                                rtts = append(rtts, n)
                        }
                        if len(rtts) > 0 {
                                // Trim 0 that are at the end.
                                stop := -1
                                for i := len(rtts) - 1; i >= 0; i-- {
                                        if rtts[i] != 0 {
                                                stop = i
                                                break
                                        }
                                }
                                rtts = rtts[:stop+1]
                        }
                        if len(rtts) > 4 {
                                // There should be at most values for "uncompressed", "fast",
                                // "better" and "best" (when some 0 are present).
                                return fmt.Errorf("compression mode %q should have no more than 4 RTT thresholds: %v", c.Mode, c.RTTThresholds)
                        } else if len(rtts) == 0 {
                                // But there should be at least 1 if the user provided the slice.
                                // We would be here only if it was provided by say with values
                                // being a single or all zeros.
                                return fmt.Errorf("compression mode %q requires at least one RTT threshold", c.Mode)
                        }
                }
                c.Mode = CompressionS2Auto
                c.RTTThresholds = rtts
        case "fast", "s2_fast":
                c.Mode = CompressionS2Fast
        case "better", "s2_better":
                c.Mode = CompressionS2Better
        case "best", "s2_best":
                c.Mode = CompressionS2Best
        default:
                return fmt.Errorf("unsupported compression mode %q", c.Mode)
        }
        return nil
}

// Returns `true` if the compression mode `m` indicates that the server
// will negotiate compression with the remote server, `false` otherwise.
// Note that the provided compression mode is assumed to have been
// normalized and validated.
func needsCompression(m string) bool {
        return m != _EMPTY_ && m != CompressionOff && m != CompressionNotSupported
}

// Compression is asymmetric, meaning that one side can have a different
// compression level than the other. However, we need to check for cases
// when this server `scm` or the remote `rcm` do not support compression
// (say older server, or test to make it behave as it is not), or have
// the compression off.
// Note that `scm` is assumed to not be "off" or "not supported".
func selectCompressionMode(scm, rcm string) (mode string, err error) {
        if rcm == CompressionNotSupported || rcm == _EMPTY_ {
                return CompressionNotSupported, nil
        }
        switch rcm {
        case CompressionOff:
                // If the remote explicitly disables compression, then we won't
                // use compression.
                return CompressionOff, nil
        case CompressionAccept:
                // If the remote is ok with compression (but is not initiating it),
                // and if we too are in this mode, then it means no compression.
                if scm == CompressionAccept {
                        return CompressionOff, nil
                }
                // Otherwise use our compression mode.
                return scm, nil
        case CompressionS2Auto, CompressionS2Uncompressed, CompressionS2Fast, CompressionS2Better, CompressionS2Best:
                // This case is here to make sure that if we don't recognize a
                // compression setting, we error out.
                if scm == CompressionAccept {
                        // If our compression mode is "accept", then we will use the remote
                        // compression mode, except if it is "auto", in which case we will
                        // default to "fast". This is not a configuration (auto in one
                        // side and accept in the other) that would be recommended.
                        if rcm == CompressionS2Auto {
                                return CompressionS2Fast, nil
                        }
                        // Use their compression mode.
                        return rcm, nil
                }
                // Otherwise use our compression mode.
                return scm, nil
        default:
                return _EMPTY_, fmt.Errorf("unsupported route compression mode %q", rcm)
        }
}

// If the configured compression mode is "auto" then will return that,
// otherwise will return the given `cm` compression mode.
func compressionModeForInfoProtocol(co *CompressionOpts, cm string) string {
        if co.Mode == CompressionS2Auto {
                return CompressionS2Auto
        }
        return cm
}

// Given a connection RTT and a list of thresholds durations, this
// function will return an S2 compression level such as "uncompressed",
// "fast", "better" or "best". For instance, with the following slice:
// [5ms, 10ms, 15ms, 20ms], a RTT of up to 5ms will result
// in the compression level "uncompressed", ]5ms..10ms] will result in
// "fast" compression, etc..
// However, the 0 value allows for disabling of some compression levels.
// For instance, the following slice: [0, 0, 20, 30] means that a RTT of
// [0..20ms] would result in the "better" compression - effectively disabling
// the use of "uncompressed" and "fast", then anything above 20ms would
// result in the use of "best" level (the 30 in the list has no effect
// and the list could have been simplified to [0, 0, 20]).
func selectS2AutoModeBasedOnRTT(rtt time.Duration, rttThresholds []time.Duration) string {
        var idx int
        var found bool
        for i, d := range rttThresholds {
                if rtt <= d {
                        idx = i
                        found = true
                        break
                }
        }
        if !found {
                // If we did not find but we have all levels, then use "best",
                // otherwise use the last one in array.
                if l := len(rttThresholds); l >= 3 {
                        idx = 3
                } else {
                        idx = l - 1
                }
        }
        switch idx {
        case 0:
                return CompressionS2Uncompressed
        case 1:
                return CompressionS2Fast
        case 2:
                return CompressionS2Better
        }
        return CompressionS2Best
}

func compressOptsEqual(c1, c2 *CompressionOpts) bool {
        if c1 == c2 {
                return true
        }
        if (c1 == nil && c2 != nil) || (c1 != nil && c2 == nil) {
                return false
        }
        if c1.Mode != c2.Mode {
                return false
        }
        // For s2_auto, if one has an empty RTTThresholds, it is equivalent
        // to the defaultCompressionS2AutoRTTThresholds array, so compare with that.
        if c1.Mode == CompressionS2Auto {
                if len(c1.RTTThresholds) == 0 && !reflect.DeepEqual(c2.RTTThresholds, defaultCompressionS2AutoRTTThresholds) {
                        return false
                }
                if len(c2.RTTThresholds) == 0 && !reflect.DeepEqual(c1.RTTThresholds, defaultCompressionS2AutoRTTThresholds) {
                        return false
                }
                if !reflect.DeepEqual(c1.RTTThresholds, c2.RTTThresholds) {
                        return false
                }
        }
        return true
}

// Returns an array of s2 WriterOption based on the route compression mode.
// So far we return a single option, but this way we can call s2.NewWriter()
// with a nil []s2.WriterOption, but not with a nil s2.WriterOption, so
// this is more versatile.
func s2WriterOptions(cm string) []s2.WriterOption {
        _opts := [2]s2.WriterOption{}
        opts := append(
                _opts[:0],
                s2.WriterConcurrency(1), // Stop asynchronous flushing in separate goroutines
        )
        switch cm {
        case CompressionS2Uncompressed:
                return append(opts, s2.WriterUncompressed())
        case CompressionS2Best:
                return append(opts, s2.WriterBestCompression())
        case CompressionS2Better:
                return append(opts, s2.WriterBetterCompression())
        default:
                return nil
        }
}

// New will setup a new server struct after parsing the options.
// DEPRECATED: Use NewServer(opts)
func New(opts *Options) *Server {
        s, _ := NewServer(opts)
        return s
}

func NewServerFromConfig(opts *Options) (*Server, error) {
        if opts.ConfigFile != _EMPTY_ && opts.configDigest == "" {
                if err := opts.ProcessConfigFile(opts.ConfigFile); err != nil {
                        return nil, err
                }
        }
        return NewServer(opts)
}

// NewServer will setup a new server struct after parsing the options.
// Could return an error if options can not be validated.
// The provided Options type should not be re-used afterwards.
// Either use Options.Clone() to pass a copy, or make a new one.
func NewServer(opts *Options) (*Server, error) {
        setBaselineOptions(opts)

        // Process TLS options, including whether we require client certificates.
        tlsReq := opts.TLSConfig != nil
        verify := (tlsReq && opts.TLSConfig.ClientAuth == tls.RequireAndVerifyClientCert)

        // Create our server's nkey identity.
        kp, _ := nkeys.CreateServer()
        pub, _ := kp.PublicKey()

        // Create an xkey for encrypting messages from this server.
        var xkp nkeys.KeyPair
        var xpub string
        if !fips140.Enabled() {
                xkp, _ = nkeys.CreateCurveKeys()
                xpub, _ = xkp.PublicKey()
        }

        serverName := pub
        if opts.ServerName != _EMPTY_ {
                serverName = opts.ServerName
        }

        httpBasePath := normalizeBasePath(opts.HTTPBasePath)

        // Validate some options. This is here because we cannot assume that
        // server will always be started with configuration parsing (that could
        // report issues). Its options can be (incorrectly) set by hand when
        // server is embedded. If there is an error, return nil.
        if err := validateOptions(opts); err != nil {
                return nil, err
        }

        info := Info{
                ID:           pub,
                XKey:         xpub,
                Version:      VERSION,
                Proto:        PROTO,
                GitCommit:    gitCommit,
                GoVersion:    runtime.Version(),
                Name:         serverName,
                Host:         opts.Host,
                Port:         opts.Port,
                AuthRequired: false,
                TLSRequired:  tlsReq && !opts.AllowNonTLS,
                TLSVerify:    verify,
                MaxPayload:   opts.MaxPayload,
                JetStream:    opts.JetStream,
                Headers:      !opts.NoHeaderSupport,
                Cluster:      opts.Cluster.Name,
                Domain:       opts.JetStreamDomain,
                JSApiLevel:   JSApiLevel,
        }

        if tlsReq && !info.TLSRequired {
                info.TLSAvailable = true
        }

        now := time.Now()

        s := &Server{
                kp:                 kp,
                xkp:                xkp,
                xpub:               xpub,
                configFile:         opts.ConfigFile,
                info:               info,
                opts:               opts,
                done:               make(chan bool, 1),
                start:              now,
                configTime:         now,
                gwLeafSubs:         NewSublistWithCache(),
                httpBasePath:       httpBasePath,
                eventIds:           nuid.New(),
                routesToSelf:       make(map[string]struct{}),
                httpReqStats:       make(map[string]uint64), // Used to track HTTP requests
                rateLimitLoggingCh: make(chan time.Duration, 1),
                leafNodeEnabled:    opts.LeafNode.Port != 0 || len(opts.LeafNode.Remotes) > 0,
                syncOutSem:         make(chan struct{}, maxConcurrentSyncRequests),
        }

        // Delayed API response queue. Create regardless if JetStream is configured
        // or not (since it can be enabled/disabled with config reload, we want this
        // queue to exist at all times).
        s.delayedAPIResponses = newIPQueue[*delayedAPIResponse](s, "delayed API responses")

        // By default we'll allow account NRG.
        s.accountNRGAllowed.Store(true)

        // Fill up the maximum in flight syncRequests for this server.
        // Used in JetStream catchup semantics.
        for i := 0; i < maxConcurrentSyncRequests; i++ {
                s.syncOutSem <- struct{}{}
        }

        if opts.TLSRateLimit > 0 {
                s.connRateCounter = newRateCounter(opts.tlsConfigOpts.RateLimit)
        }

        // Trusted root operator keys.
        if !s.processTrustedKeys() {
                return nil, fmt.Errorf("Error processing trusted operator keys")
        }

        // If we have solicited leafnodes but no clustering and no clustername.
        // However we may need a stable clustername so use the server name.
        if len(opts.LeafNode.Remotes) > 0 && opts.Cluster.Port == 0 && opts.Cluster.Name == _EMPTY_ {
                s.leafNoCluster = true
                opts.Cluster.Name = opts.ServerName
        }

        if opts.Cluster.Name != _EMPTY_ {
                // Also place into mapping cn with cnMu lock.
                s.cnMu.Lock()
                s.cn = opts.Cluster.Name
                s.cnMu.Unlock()
        }

        s.mu.Lock()
        defer s.mu.Unlock()

        // If there are proxies trusted public keys in the configuration
        // this will fill create the corresponding list of nkeys.KeyPair
        // that we can use for signature verification.
        s.processProxiesTrustedKeys()

        // Place ourselves in the JetStream nodeInfo if needed.
        if opts.JetStream {
                ourNode := getHash(serverName)
                s.nodeToInfo.Store(ourNode, nodeInfo{
                        name:            serverName,
                        version:         VERSION,
                        cluster:         opts.Cluster.Name,
                        domain:          opts.JetStreamDomain,
                        id:              info.ID,
                        tags:            opts.Tags,
                        cfg:             &JetStreamConfig{MaxMemory: opts.JetStreamMaxMemory, MaxStore: opts.JetStreamMaxStore, CompressOK: true},
                        stats:           nil,
                        offline:         false,
                        js:              true,
                        binarySnapshots: true,
                        accountNRG:      true,
                })
        }

        s.routeResolver = opts.Cluster.resolver
        if s.routeResolver == nil {
                s.routeResolver = net.DefaultResolver
        }

        // Used internally for quick look-ups.
        s.clientConnectURLsMap = make(refCountedUrlSet)
        s.websocket.connectURLsMap = make(refCountedUrlSet)
        s.leafURLsMap = make(refCountedUrlSet)

        // Ensure that non-exported options (used in tests) are properly set.
        s.setLeafNodeNonExportedOptions()

        // Setup OCSP Stapling and OCSP Peer. This will abort server from starting if there
        // are no valid staples and OCSP Stapling policy is set to Always or MustStaple.
        if err := s.enableOCSP(); err != nil {
                return nil, err
        }

        // Call this even if there is no gateway defined. It will
        // initialize the structure so we don't have to check for
        // it to be nil or not in various places in the code.
        if err := s.newGateway(opts); err != nil {
                return nil, err
        }

        // If we have a cluster definition but do not have a cluster name, create one.
        if opts.Cluster.Port != 0 && opts.Cluster.Name == _EMPTY_ {
                s.info.Cluster = nuid.Next()
        } else if opts.Cluster.Name != _EMPTY_ {
                // Likewise here if we have a cluster name set.
                s.info.Cluster = opts.Cluster.Name
        }

        // This is normally done in the AcceptLoop, once the
        // listener has been created (possibly with random port),
        // but since some tests may expect the INFO to be properly
        // set after New(), let's do it now.
        s.setInfoHostPort()

        // For tracking clients
        s.clients = make(map[uint64]*client)

        // For tracking closed clients.
        s.closed = newClosedRingBuffer(opts.MaxClosedClients)

        // For tracking connections that are not yet registered
        // in s.routes, but for which readLoop has started.
        s.grTmpClients = make(map[uint64]*client)

        // For tracking routes and their remote ids
        s.initRouteStructures(opts)

        // For tracking leaf nodes.
        s.leafs = make(map[uint64]*client)

        // Used to kick out all go routines possibly waiting on server
        // to shutdown.
        s.quitCh = make(chan struct{})

        // Closed when startup is complete. ReadyForConnections() will block on
        // this before checking the presence of listening sockets.
        s.startupComplete = make(chan struct{})

        // Closed when Shutdown() is complete. Allows WaitForShutdown() to block
        // waiting for complete shutdown.
        s.shutdownComplete = make(chan struct{})

        // Check for configured account resolvers.
        if err := s.configureResolver(); err != nil {
                return nil, err
        }
        // If there is an URL account resolver, do basic test to see if anyone is home.
        if ar := opts.AccountResolver; ar != nil {
                if ur, ok := ar.(*URLAccResolver); ok {
                        if _, err := ur.Fetch(_EMPTY_); err != nil {
                                return nil, err
                        }
                }
        }
        // For other resolver:
        // In operator mode, when the account resolver depends on an external system and
        // the system account can't be fetched, inject a temporary one.
        if ar := s.accResolver; len(opts.TrustedOperators) == 1 && ar != nil &&
                opts.SystemAccount != _EMPTY_ && opts.SystemAccount != DEFAULT_SYSTEM_ACCOUNT {
                if _, ok := ar.(*MemAccResolver); !ok {
                        s.mu.Unlock()
                        var a *Account
                        // perform direct lookup to avoid warning trace
                        if _, err := fetchAccount(ar, opts.SystemAccount); err == nil {
                                a, _ = s.lookupAccount(opts.SystemAccount)
                        }
                        s.mu.Lock()
                        if a == nil {
                                sac := NewAccount(opts.SystemAccount)
                                sac.Issuer = opts.TrustedOperators[0].Issuer
                                sac.signingKeys = map[string]jwt.Scope{}
                                sac.signingKeys[opts.SystemAccount] = nil
                                s.registerAccountNoLock(sac)
                        }
                }
        }

        // For tracking accounts
        if _, err := s.configureAccounts(false); err != nil {
                return nil, err
        }

        // Used to setup Authorization.
        s.configureAuthorization()

        // Start signal handler
        s.handleSignals()

        return s, nil
}

// Initializes route structures based on pooling and/or per-account routes.
//
// Server lock is held on entry
func (s *Server) initRouteStructures(opts *Options) {
        s.routes = make(map[string][]*client)
        if ps := opts.Cluster.PoolSize; ps > 0 {
                s.routesPoolSize = ps
        } else {
                s.routesPoolSize = 1
        }
        // If we have per-account routes, we create accRoutes and initialize it
        // with nil values. The presence of an account as the key will allow us
        // to know if a given account is supposed to have dedicated routes.
        if l := len(opts.Cluster.PinnedAccounts); l > 0 {
                s.accRoutes = make(map[string]map[string]*client, l)
                for _, acc := range opts.Cluster.PinnedAccounts {
                        s.accRoutes[acc] = make(map[string]*client)
                }
        }
}

func (s *Server) logRejectedTLSConns() {
        defer s.grWG.Done()
        t := time.NewTicker(time.Second)
        defer t.Stop()
        for {
                select {
                case <-s.quitCh:
                        return
                case <-t.C:
                        blocked := s.connRateCounter.countBlocked()
                        if blocked > 0 {
                                s.Warnf("Rejected %d connections due to TLS rate limiting", blocked)
                        }
                }
        }
}

// clusterName returns our cluster name which could be dynamic.
func (s *Server) ClusterName() string {
        s.mu.RLock()
        cn := s.info.Cluster
        s.mu.RUnlock()
        return cn
}

// Grabs cluster name with cluster name specific lock.
func (s *Server) cachedClusterName() string {
        s.cnMu.RLock()
        cn := s.cn
        s.cnMu.RUnlock()
        return cn
}

// setClusterName will update the cluster name for this server.
func (s *Server) setClusterName(name string) {
        s.mu.Lock()
        var resetCh chan struct{}
        if s.sys != nil && s.info.Cluster != name {
                // can't hold the lock as go routine reading it may be waiting for lock as well
                resetCh = s.sys.resetCh
        }
        s.info.Cluster = name
        s.routeInfo.Cluster = name

        // Need to close solicited leaf nodes. The close has to be done outside of the server lock.
        var leafs []*client
        for _, c := range s.leafs {
                c.mu.Lock()
                if c.leaf != nil && c.leaf.remote != nil {
                        leafs = append(leafs, c)
                }
                c.mu.Unlock()
        }
        s.mu.Unlock()

        // Also place into mapping cn with cnMu lock.
        s.cnMu.Lock()
        s.cn = name
        s.cnMu.Unlock()

        for _, l := range leafs {
                l.closeConnection(ClusterNameConflict)
        }
        if resetCh != nil {
                resetCh <- struct{}{}
        }
        s.Noticef("Cluster name updated to %s", name)
}

// Return whether the cluster name is dynamic.
func (s *Server) isClusterNameDynamic() bool {
        // We need to lock the whole "Cluster.Name" check and not use s.getOpts()
        // because otherwise this could cause a data race with setting the name in
        // route.go's processRouteConnect().
        s.optsMu.RLock()
        dynamic := s.opts.Cluster.Name == _EMPTY_
        s.optsMu.RUnlock()
        return dynamic
}

// Returns our configured serverName.
func (s *Server) serverName() string {
        return s.getOpts().ServerName
}

// ClientURL returns the URL used to connect clients.
// Helpful in tests and with in-process servers using a random client port (-1).
func (s *Server) ClientURL() string {
        // FIXME(dlc) - should we add in user and pass if defined single?
        opts := s.getOpts()
        var u url.URL
        u.Scheme = "nats"
        if opts.TLSConfig != nil {
                u.Scheme = "tls"
        }
        u.Host = net.JoinHostPort(opts.Host, fmt.Sprintf("%d", opts.Port))
        return u.String()
}

// WebsocketURL returns the URL used to connect websocket clients.
// Helpful in tests and with in-process servers using a random websocket port (-1).
func (s *Server) WebsocketURL() string {
        opts := s.getOpts()
        var u url.URL
        u.Scheme = "ws"
        if opts.Websocket.TLSConfig != nil {
                u.Scheme = "wss"
        }
        u.Host = net.JoinHostPort(opts.Websocket.Host, fmt.Sprintf("%d", opts.Websocket.Port))
        return u.String()
}

func validateCluster(o *Options) error {
        if o.Cluster.Name != _EMPTY_ && strings.Contains(o.Cluster.Name, " ") {
                return ErrClusterNameHasSpaces
        }
        if o.Cluster.Compression.Mode != _EMPTY_ {
                if err := validateAndNormalizeCompressionOption(&o.Cluster.Compression, CompressionS2Fast); err != nil {
                        return err
                }
        }
        if err := validatePinnedCerts(o.Cluster.TLSPinnedCerts); err != nil {
                return fmt.Errorf("cluster: %v", err)
        }
        // Check that cluster name if defined matches any gateway name.
        // Note that we have already verified that the gateway name does not have spaces.
        if o.Gateway.Name != _EMPTY_ && o.Gateway.Name != o.Cluster.Name {
                if o.Cluster.Name != _EMPTY_ {
                        return ErrClusterNameConfigConflict
                }
                // Set this here so we do not consider it dynamic.
                o.Cluster.Name = o.Gateway.Name
        }
        if l := len(o.Cluster.PinnedAccounts); l > 0 {
                if o.Cluster.PoolSize < 0 {
                        return fmt.Errorf("pool_size cannot be negative if pinned accounts are specified")
                }
                m := make(map[string]struct{}, l)
                for _, a := range o.Cluster.PinnedAccounts {
                        if _, exists := m[a]; exists {
                                return fmt.Errorf("found duplicate account name %q in pinned accounts list %q", a, o.Cluster.PinnedAccounts)
                        }
                        m[a] = struct{}{}
                }
        }
        return nil
}

func validatePinnedCerts(pinned PinnedCertSet) error {
        re := regexp.MustCompile("^[a-f0-9]{64}$")
        for certId := range pinned {
                entry := strings.ToLower(certId)
                if !re.MatchString(entry) {
                        return fmt.Errorf("error parsing 'pinned_certs' key %s does not look like lower case hex-encoded sha256 of DER encoded SubjectPublicKeyInfo", entry)
                }
        }
        return nil
}

func validateOptions(o *Options) error {
        if o.LameDuckDuration > 0 && o.LameDuckGracePeriod >= o.LameDuckDuration {
                return fmt.Errorf("lame duck grace period (%v) should be strictly lower than lame duck duration (%v)",
                        o.LameDuckGracePeriod, o.LameDuckDuration)
        }
        if int64(o.MaxPayload) > o.MaxPending {
                return fmt.Errorf("max_payload (%v) cannot be higher than max_pending (%v)",
                        o.MaxPayload, o.MaxPending)
        }
        if o.ServerName != _EMPTY_ && strings.Contains(o.ServerName, " ") {
                return errors.New("server name cannot contain spaces")
        }
        // Check that the trust configuration is correct.
        if err := validateTrustedOperators(o); err != nil {
                return err
        }
        // Check on leaf nodes which will require a system
        // account when gateways are also configured.
        if err := validateLeafNode(o); err != nil {
                return err
        }
        // Check that authentication is properly configured.
        if err := validateAuth(o); err != nil {
                return err
        }
        // Check that proxies is properly configured.
        if err := validateProxies(o); err != nil {
                return err
        }
        // Check that gateway is properly configured. Returns no error
        // if there is no gateway defined.
        if err := validateGatewayOptions(o); err != nil {
                return err
        }
        // Check that cluster name if defined matches any gateway name.
        if err := validateCluster(o); err != nil {
                return err
        }
        if err := validateMQTTOptions(o); err != nil {
                return err
        }
        if err := validateJetStreamOptions(o); err != nil {
                return err
        }
        // Finally check websocket options.
        return validateWebsocketOptions(o)
}

func (s *Server) getOpts() *Options {
        s.optsMu.RLock()
        opts := s.opts
        s.optsMu.RUnlock()
        return opts
}

func (s *Server) setOpts(opts *Options) {
        s.optsMu.Lock()
        s.opts = opts
        s.optsMu.Unlock()
}

func (s *Server) globalAccount() *Account {
        s.mu.RLock()
        gacc := s.gacc
        s.mu.RUnlock()
        return gacc
}

// Used to setup or update Accounts.
// Returns a map that indicates which accounts have had their stream imports
// changed (in case of an update in configuration reload).
// Lock is held upon entry, but will be released/reacquired in this function.
func (s *Server) configureAccounts(reloading bool) (map[string]struct{}, error) {
        awcsti := make(map[string]struct{})

        // Create the global account.
        if s.gacc == nil {
                s.gacc = NewAccount(globalAccountName)
                s.registerAccountNoLock(s.gacc)
        }

        opts := s.getOpts()

        // We need to track service imports since we can not swap them out (unsub and re-sub)
        // until the proper server struct accounts have been swapped in properly. Doing it in
        // place could lead to data loss or server panic since account under new si has no real
        // account and hence no sublist, so will panic on inbound message.
        siMap := make(map[*Account][][]byte)

        // Check opts and walk through them. We need to copy them here
        // so that we do not keep a real one sitting in the options.
        for _, acc := range opts.Accounts {
                var a *Account
                create := true
                // For the global account, we want to skip the reload process
                // and fall back into the "create" case which will in that
                // case really be just an update (shallowCopy will make sure
                // that mappings are copied over).
                if reloading && acc.Name != globalAccountName {
                        if ai, ok := s.accounts.Load(acc.Name); ok {
                                a = ai.(*Account)
                                // Before updating the account, check if stream imports have changed.
                                if !a.checkStreamImportsEqual(acc) {
                                        awcsti[acc.Name] = struct{}{}
                                }
                                a.mu.Lock()
                                // Collect the sids for the service imports since we are going to
                                // replace with new ones.
                                var sids [][]byte
                                for _, sis := range a.imports.services {
                                        for _, si := range sis {
                                                if si.sid != nil {
                                                        sids = append(sids, si.sid)
                                                }
                                        }
                                }
                                // Setup to process later if needed.
                                if len(sids) > 0 || len(acc.imports.services) > 0 {
                                        siMap[a] = sids
                                }

                                // Now reset all export/imports fields since they are going to be
                                // filled in shallowCopy()
                                a.imports.streams, a.imports.services = nil, nil
                                a.exports.streams, a.exports.services = nil, nil
                                // We call shallowCopy from the account `acc` (the one in Options)
                                // and pass `a` (our existing account) to get it updated.
                                acc.shallowCopy(a)
                                a.mu.Unlock()
                                create = false
                        }
                }
                // Track old mappings if global account.
                var oldGMappings []*mapping
                if create {
                        if acc.Name == globalAccountName {
                                a = s.gacc
                                a.mu.Lock()
                                oldGMappings = append(oldGMappings, a.mappings...)
                                a.mu.Unlock()
                        } else {
                                a = NewAccount(acc.Name)
                        }
                        // Locking matters in the case of an update of the global account
                        a.mu.Lock()
                        acc.shallowCopy(a)
                        a.mu.Unlock()
                        // Will be a no-op in case of the global account since it is already registered.
                        s.registerAccountNoLock(a)
                }

                // The `acc` account is stored in options, not in the server, and these can be cleared.
                acc.sl, acc.clients, acc.mappings = nil, nil, nil

                // Check here if we have been reloaded and we have a global account with mappings that may have changed.
                // If we have leafnodes they need to be updated.
                if reloading && a == s.gacc {
                        a.mu.Lock()
                        mappings := make(map[string]*mapping)
                        if len(a.mappings) > 0 && a.nleafs > 0 {
                                for _, em := range a.mappings {
                                        mappings[em.src] = em
                                }
                        }
                        a.mu.Unlock()
                        if len(mappings) > 0 || len(oldGMappings) > 0 {
                                a.lmu.RLock()
                                for _, lc := range a.lleafs {
                                        for _, em := range mappings {
                                                lc.forceAddToSmap(em.src)
                                        }
                                        // Remove any old ones if needed.
                                        for _, em := range oldGMappings {
                                                // Only remove if not in the new ones.
                                                if _, ok := mappings[em.src]; !ok {
                                                        lc.forceRemoveFromSmap(em.src)
                                                }
                                        }
                                }
                                a.lmu.RUnlock()
                        }
                }

                // If we see an account defined using $SYS we will make sure that is set as system account.
                if acc.Name == DEFAULT_SYSTEM_ACCOUNT && opts.SystemAccount == _EMPTY_ {
                        opts.SystemAccount = DEFAULT_SYSTEM_ACCOUNT
                }
        }

        // Now that we have this we need to remap any referenced accounts in
        // import or export maps to the new ones.
        swapApproved := func(ea *exportAuth) {
                for sub, a := range ea.approved {
                        var acc *Account
                        if v, ok := s.accounts.Load(a.Name); ok {
                                acc = v.(*Account)
                        }
                        ea.approved[sub] = acc
                }
        }
        var numAccounts int
        s.accounts.Range(func(k, v any) bool {
                numAccounts++
                acc := v.(*Account)
                acc.mu.Lock()
                // Exports
                for _, se := range acc.exports.streams {
                        if se != nil {
                                swapApproved(&se.exportAuth)
                        }
                }
                for _, se := range acc.exports.services {
                        if se != nil {
                                // Swap over the bound account for service exports.
                                if se.acc != nil {
                                        if v, ok := s.accounts.Load(se.acc.Name); ok {
                                                se.acc = v.(*Account)
                                        }
                                }
                                swapApproved(&se.exportAuth)
                        }
                }
                // Imports
                for _, si := range acc.imports.streams {
                        if v, ok := s.accounts.Load(si.acc.Name); ok {
                                si.acc = v.(*Account)
                        }
                }
                for _, sis := range acc.imports.services {
                        for _, si := range sis {
                                if v, ok := s.accounts.Load(si.acc.Name); ok {
                                        si.acc = v.(*Account)

                                        // It is possible to allow for latency tracking inside your
                                        // own account, so lock only when not the same account.
                                        if si.acc == acc {
                                                si.se = si.acc.getServiceExport(si.to)
                                                continue
                                        }
                                        si.acc.mu.RLock()
                                        si.se = si.acc.getServiceExport(si.to)
                                        si.acc.mu.RUnlock()
                                }
                        }
                }
                // Make sure the subs are running, but only if not reloading.
                if len(acc.imports.services) > 0 && acc.ic == nil && !reloading {
                        acc.ic = s.createInternalAccountClient()
                        acc.ic.acc = acc
                        // Need to release locks to invoke this function.
                        acc.mu.Unlock()
                        s.mu.Unlock()
                        acc.addAllServiceImportSubs()
                        s.mu.Lock()
                        acc.mu.Lock()
                }
                acc.updated = time.Now()
                acc.mu.Unlock()
                return true
        })

        // Check if we need to process service imports pending from above.
        // This processing needs to be after we swap in the real accounts above.
        for acc, sids := range siMap {
                c := acc.ic
                for _, sid := range sids {
                        c.processUnsub(sid)
                }
                acc.addAllServiceImportSubs()
                s.mu.Unlock()
                s.registerSystemImports(acc)
                s.mu.Lock()
        }

        // Set the system account if it was configured.
        // Otherwise create a default one.
        if opts.SystemAccount != _EMPTY_ {
                // Lock may be acquired in lookupAccount, so release to call lookupAccount.
                s.mu.Unlock()
                acc, err := s.lookupAccount(opts.SystemAccount)
                s.mu.Lock()
                if err == nil && s.sys != nil && acc != s.sys.account {
                        // sys.account.clients (including internal client)/respmap/etc... are transferred separately
                        s.sys.account = acc
                        s.sysAcc.Store(acc)
                }
                if err != nil {
                        return awcsti, fmt.Errorf("error resolving system account: %v", err)
                }

                // If we have defined a system account here check to see if its just us and the $G account.
                // We would do this to add user/pass to the system account. If this is the case add in
                // no-auth-user for $G.
                // Only do this if non-operator mode and we did not have an authorization block defined.
                if len(opts.TrustedOperators) == 0 && numAccounts == 2 && opts.NoAuthUser == _EMPTY_ && !opts.authBlockDefined {
                        // If we come here from config reload, let's not recreate the fake user name otherwise
                        // it will cause currently clients to be disconnected.
                        uname := s.sysAccOnlyNoAuthUser
                        if uname == _EMPTY_ {
                                // Create a unique name so we do not collide.
                                var b [8]byte
                                rn := rand.Int63()
                                for i, l := 0, rn; i < len(b); i++ {
                                        b[i] = digits[l%base]
                                        l /= base
                                }
                                uname = fmt.Sprintf("nats-%s", b[:])
                                s.sysAccOnlyNoAuthUser = uname
                        }
                        opts.Users = append(opts.Users, &User{Username: uname, Password: uname[6:], Account: s.gacc})
                        opts.NoAuthUser = uname
                }
        }

        // Add any required exports from system account.
        if s.sys != nil {
                sysAcc := s.sys.account
                s.mu.Unlock()
                s.addSystemAccountExports(sysAcc)
                s.mu.Lock()
        }

        return awcsti, nil
}

// Setup the account resolver. For memory resolver, make sure the JWTs are
// properly formed but do not enforce expiration etc.
// Lock is held on entry, but may be released/reacquired during this call.
func (s *Server) configureResolver() error {
        opts := s.getOpts()
        s.accResolver = opts.AccountResolver
        if opts.AccountResolver != nil {
                // For URL resolver, set the TLSConfig if specified.
                if opts.AccountResolverTLSConfig != nil {
                        if ar, ok := opts.AccountResolver.(*URLAccResolver); ok {
                                if t, ok := ar.c.Transport.(*http.Transport); ok {
                                        t.CloseIdleConnections()
                                        t.TLSClientConfig = opts.AccountResolverTLSConfig.Clone()
                                }
                        }
                }
                if len(opts.resolverPreloads) > 0 {
                        // Lock ordering is account resolver -> server, so we need to release
                        // the lock and reacquire it when done with account resolver's calls.
                        ar := s.accResolver
                        s.mu.Unlock()
                        defer s.mu.Lock()
                        if ar.IsReadOnly() {
                                return fmt.Errorf("resolver preloads only available for writeable resolver types MEM/DIR/CACHE_DIR")
                        }
                        for k, v := range opts.resolverPreloads {
                                _, err := jwt.DecodeAccountClaims(v)
                                if err != nil {
                                        return fmt.Errorf("preload account error for %q: %v", k, err)
                                }
                                ar.Store(k, v)
                        }
                }
        }
        return nil
}

// This will check preloads for validation issues.
func (s *Server) checkResolvePreloads() {
        opts := s.getOpts()
        // We can just check the read-only opts versions here, that way we do not need
        // to grab server lock or access s.accResolver.
        for k, v := range opts.resolverPreloads {
                claims, err := jwt.DecodeAccountClaims(v)
                if err != nil {
                        s.Errorf("Preloaded account [%s] not valid", k)
                        continue
                }
                // Check if it is expired.
                vr := jwt.CreateValidationResults()
                claims.Validate(vr)
                if vr.IsBlocking(true) {
                        s.Warnf("Account [%s] has validation issues:", k)
                        for _, v := range vr.Issues {
                                s.Warnf("  - %s", v.Description)
                        }
                }
        }
}

// Determines if we are in pre NATS 2.0 setup with no accounts.
func (s *Server) globalAccountOnly() bool {
        var hasOthers bool

        if s.trustedKeys != nil {
                return false
        }

        s.mu.RLock()
        s.accounts.Range(func(k, v any) bool {
                acc := v.(*Account)
                // Ignore global and system
                if acc == s.gacc || (s.sys != nil && acc == s.sys.account) {
                        return true
                }
                hasOthers = true
                return false
        })
        s.mu.RUnlock()

        return !hasOthers
}

// Determines if this server is in standalone mode, meaning no routes or gateways.
func (s *Server) standAloneMode() bool {
        opts := s.getOpts()
        return opts.Cluster.Port == 0 && opts.Gateway.Port == 0
}

func (s *Server) configuredRoutes() int {
        return len(s.getOpts().Routes)
}

// activePeers is used in bootstrapping raft groups like the JetStream meta controller.
func (s *Server) ActivePeers() (peers []string) {
        s.nodeToInfo.Range(func(k, v any) bool {
                si := v.(nodeInfo)
                if !si.offline {
                        peers = append(peers, k.(string))
                }
                return true
        })
        return peers
}

// isTrustedIssuer will check that the issuer is a trusted public key.
// This is used to make sure an account was signed by a trusted operator.
func (s *Server) isTrustedIssuer(issuer string) bool {
        s.mu.RLock()
        defer s.mu.RUnlock()
        // If we are not running in trusted mode and there is no issuer, that is ok.
        if s.trustedKeys == nil && issuer == _EMPTY_ {
                return true
        }
        for _, tk := range s.trustedKeys {
                if tk == issuer {
                        return true
                }
        }
        return false
}

// processTrustedKeys will process binary stamped and
// options-based trusted nkeys. Returns success.
func (s *Server) processTrustedKeys() bool {
        s.strictSigningKeyUsage = map[string]struct{}{}
        opts := s.getOpts()
        if trustedKeys != _EMPTY_ && !s.initStampedTrustedKeys() {
                return false
        } else if opts.TrustedKeys != nil {
                for _, key := range opts.TrustedKeys {
                        if !nkeys.IsValidPublicOperatorKey(key) {
                                return false
                        }
                }
                s.trustedKeys = append([]string(nil), opts.TrustedKeys...)
                for _, claim := range opts.TrustedOperators {
                        if !claim.StrictSigningKeyUsage {
                                continue
                        }
                        for _, key := range claim.SigningKeys {
                                s.strictSigningKeyUsage[key] = struct{}{}
                        }
                }
        }
        return true
}

// checkTrustedKeyString will check that the string is a valid array
// of public operator nkeys.
func checkTrustedKeyString(keys string) []string {
        tks := strings.Fields(keys)
        if len(tks) == 0 {
                return nil
        }
        // Walk all the keys and make sure they are valid.
        for _, key := range tks {
                if !nkeys.IsValidPublicOperatorKey(key) {
                        return nil
                }
        }
        return tks
}

// initStampedTrustedKeys will check the stamped trusted keys
// and will set the server field 'trustedKeys'. Returns whether
// it succeeded or not.
func (s *Server) initStampedTrustedKeys() bool {
        // Check to see if we have an override in options, which will cause us to fail.
        if len(s.getOpts().TrustedKeys) > 0 {
                return false
        }
        tks := checkTrustedKeyString(trustedKeys)
        if len(tks) == 0 {
                return false
        }
        s.trustedKeys = tks
        return true
}

// PrintAndDie is exported for access in other packages.
func PrintAndDie(msg string) {
        fmt.Fprintln(os.Stderr, msg)
        os.Exit(1)
}

// PrintServerAndExit will print our version and exit.
func PrintServerAndExit() {
        fmt.Printf("nats-server: v%s\n", VERSION)
        os.Exit(0)
}

// ProcessCommandLineArgs takes the command line arguments
// validating and setting flags for handling in case any
// sub command was present.
func ProcessCommandLineArgs(cmd *flag.FlagSet) (showVersion bool, showHelp bool, err error) {
        if len(cmd.Args()) > 0 {
                arg := cmd.Args()[0]
                switch strings.ToLower(arg) {
                case "version":
                        return true, false, nil
                case "help":
                        return false, true, nil
                default:
                        return false, false, fmt.Errorf("unrecognized command: %q", arg)
                }
        }

        return false, false, nil
}

// Public version.
func (s *Server) Running() bool {
        return s.isRunning()
}

// Protected check on running state
func (s *Server) isRunning() bool {
        return s.running.Load()
}

func (s *Server) logPid() error {
        pidStr := strconv.Itoa(os.Getpid())
        return os.WriteFile(s.getOpts().PidFile, []byte(pidStr), defaultFilePerms)
}

// numReservedAccounts will return the number of reserved accounts configured in the server.
// Currently this is 1, one for the global default account.
func (s *Server) numReservedAccounts() int {
        return 1
}

// NumActiveAccounts reports number of active accounts on this server.
func (s *Server) NumActiveAccounts() int32 {
        return atomic.LoadInt32(&s.activeAccounts)
}

// incActiveAccounts() just adds one under lock.
func (s *Server) incActiveAccounts() {
        atomic.AddInt32(&s.activeAccounts, 1)
}

// decActiveAccounts() just subtracts one under lock.
func (s *Server) decActiveAccounts() {
        atomic.AddInt32(&s.activeAccounts, -1)
}

// This should be used for testing only. Will be slow since we have to
// range over all accounts in the sync.Map to count.
func (s *Server) numAccounts() int {
        count := 0
        s.mu.RLock()
        s.accounts.Range(func(k, v any) bool {
                count++
                return true
        })
        s.mu.RUnlock()
        return count
}

// NumLoadedAccounts returns the number of loaded accounts.
func (s *Server) NumLoadedAccounts() int {
        return s.numAccounts()
}

// LookupOrRegisterAccount will return the given account if known or create a new entry.
func (s *Server) LookupOrRegisterAccount(name string) (account *Account, isNew bool) {
        s.mu.Lock()
        defer s.mu.Unlock()
        if v, ok := s.accounts.Load(name); ok {
                return v.(*Account), false
        }
        acc := NewAccount(name)
        s.registerAccountNoLock(acc)
        return acc, true
}

// RegisterAccount will register an account. The account must be new
// or this call will fail.
func (s *Server) RegisterAccount(name string) (*Account, error) {
        s.mu.Lock()
        defer s.mu.Unlock()
        if _, ok := s.accounts.Load(name); ok {
                return nil, ErrAccountExists
        }
        acc := NewAccount(name)
        s.registerAccountNoLock(acc)
        return acc, nil
}

// SetSystemAccount will set the internal system account.
// If root operators are present it will also check validity.
func (s *Server) SetSystemAccount(accName string) error {
        // Lookup from sync.Map first.
        if v, ok := s.accounts.Load(accName); ok {
                return s.setSystemAccount(v.(*Account))
        }

        // If we are here we do not have local knowledge of this account.
        // Do this one by hand to return more useful error.
        ac, jwt, err := s.fetchAccountClaims(accName)
        if err != nil {
                return err
        }
        acc := s.buildInternalAccount(ac)
        acc.claimJWT = jwt
        // Due to race, we need to make sure that we are not
        // registering twice.
        if racc := s.registerAccount(acc); racc != nil {
                return nil
        }
        return s.setSystemAccount(acc)
}

// SystemAccount returns the system account if set.
func (s *Server) SystemAccount() *Account {
        return s.sysAcc.Load()
}

// GlobalAccount returns the global account.
// Default clients will use the global account.
func (s *Server) GlobalAccount() *Account {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.gacc
}

// SetDefaultSystemAccount will create a default system account if one is not present.
func (s *Server) SetDefaultSystemAccount() error {
        if _, isNew := s.LookupOrRegisterAccount(DEFAULT_SYSTEM_ACCOUNT); !isNew {
                return nil
        }
        s.Debugf("Created system account: %q", DEFAULT_SYSTEM_ACCOUNT)
        return s.SetSystemAccount(DEFAULT_SYSTEM_ACCOUNT)
}

// Assign a system account. Should only be called once.
// This sets up a server to send and receive messages from
// inside the server itself.
func (s *Server) setSystemAccount(acc *Account) error {
        if acc == nil {
                return ErrMissingAccount
        }
        // Don't try to fix this here.
        if acc.IsExpired() {
                return ErrAccountExpired
        }
        // If we are running with trusted keys for an operator
        // make sure we check the account is legit.
        if !s.isTrustedIssuer(acc.Issuer) {
                return ErrAccountValidation
        }

        s.mu.Lock()

        if s.sys != nil {
                s.mu.Unlock()
                return ErrAccountExists
        }

        // This is here in an attempt to quiet the race detector and not have to place
        // locks on fast path for inbound messages and checking service imports.
        acc.mu.Lock()
        if acc.imports.services == nil {
                acc.imports.services = make(map[string][]*serviceImport)
        }
        acc.mu.Unlock()

        s.sys = &internal{
                account: acc,
                client:  s.createInternalSystemClient(),
                seq:     1,
                sid:     1,
                servers: make(map[string]*serverUpdate),
                replies: make(map[string]msgHandler),
                sendq:   newIPQueue[*pubMsg](s, "System sendQ"),
                recvq:   newIPQueue[*inSysMsg](s, "System recvQ"),
                recvqp:  newIPQueue[*inSysMsg](s, "System recvQ Pings"),
                resetCh: make(chan struct{}),
                sq:      s.newSendQ(acc),
                statsz:  statsHBInterval,
                orphMax: 5 * eventsHBInterval,
                chkOrph: 3 * eventsHBInterval,
        }
        recvq, recvqp := s.sys.recvq, s.sys.recvqp
        s.sys.wg.Add(1)
        s.mu.Unlock()

        // Store in atomic for fast lookup.
        s.sysAcc.Store(acc)

        // Register with the account.
        s.sys.client.registerWithAccount(acc)

        s.addSystemAccountExports(acc)

        // Start our internal loop to serialize outbound messages.
        // We do our own wg here since we will stop first during shutdown.
        go s.internalSendLoop(&s.sys.wg)

        // Start the internal loop for inbound messages.
        go s.internalReceiveLoop(recvq)
        // Start the internal loop for inbound STATSZ/Ping messages.
        go s.internalReceiveLoop(recvqp)

        // Start up our general subscriptions
        s.initEventTracking()

        // Track for dead remote servers.
        s.wrapChk(s.startRemoteServerSweepTimer)()

        // Send out statsz updates periodically.
        s.wrapChk(s.startStatszTimer)()

        // If we have existing accounts make sure we enable account tracking.
        s.mu.Lock()
        s.accounts.Range(func(k, v any) bool {
                acc := v.(*Account)
                s.enableAccountTracking(acc)
                return true
        })
        s.mu.Unlock()

        return nil
}

// Creates an internal system client.
func (s *Server) createInternalSystemClient() *client {
        return s.createInternalClient(SYSTEM)
}

// Creates an internal jetstream client.
func (s *Server) createInternalJetStreamClient() *client {
        return s.createInternalClient(JETSTREAM)
}

// Creates an internal client for Account.
func (s *Server) createInternalAccountClient() *client {
        return s.createInternalClient(ACCOUNT)
}

// Internal clients. kind should be SYSTEM, JETSTREAM or ACCOUNT
func (s *Server) createInternalClient(kind int) *client {
        if !isInternalClient(kind) {
                return nil
        }
        now := time.Now()
        c := &client{srv: s, kind: kind, opts: internalOpts, msubs: -1, mpay: -1, start: now, last: now}
        c.initClient()
        c.echo = false
        c.headers = true
        c.flags.set(noReconnect)
        return c
}

// Determine if accounts should track subscriptions for
// efficient propagation.
// Lock should be held on entry.
func (s *Server) shouldTrackSubscriptions() bool {
        opts := s.getOpts()
        return (opts.Cluster.Port != 0 || opts.Gateway.Port != 0)
}

// Invokes registerAccountNoLock under the protection of the server lock.
// That is, server lock is acquired/released in this function.
// See registerAccountNoLock for comment on returned value.
func (s *Server) registerAccount(acc *Account) *Account {
        s.mu.Lock()
        racc := s.registerAccountNoLock(acc)
        s.mu.Unlock()
        return racc
}

// Helper to set the sublist based on preferences.
func (s *Server) setAccountSublist(acc *Account) {
        if acc != nil && acc.sl == nil {
                opts := s.getOpts()
                if opts != nil && opts.NoSublistCache {
                        acc.sl = NewSublistNoCache()
                } else {
                        acc.sl = NewSublistWithCache()
                }
        }
}

// Registers an account in the server.
// Due to some locking considerations, we may end-up trying
// to register the same account twice. This function will
// then return the already registered account.
// Lock should be held on entry.
func (s *Server) registerAccountNoLock(acc *Account) *Account {
        // We are under the server lock. Lookup from map, if present
        // return existing account.
        if a, _ := s.accounts.Load(acc.Name); a != nil {
                s.tmpAccounts.Delete(acc.Name)
                return a.(*Account)
        }
        // Finish account setup and store.
        s.setAccountSublist(acc)

        acc.mu.Lock()
        s.setRouteInfo(acc)
        if acc.clients == nil {
                acc.clients = make(map[*client]struct{})
        }

        // If we are capable of routing we will track subscription
        // information for efficient interest propagation.
        // During config reload, it is possible that account was
        // already created (global account), so use locking and
        // make sure we create only if needed.
        // TODO(dlc)- Double check that we need this for GWs.
        if acc.rm == nil && s.opts != nil && s.shouldTrackSubscriptions() {
                acc.rm = make(map[string]int32)
                acc.lqws = make(map[string]int32)
        }
        acc.srv = s
        acc.updated = time.Now()
        accName := acc.Name
        jsEnabled := len(acc.jsLimits) > 0
        acc.mu.Unlock()

        if opts := s.getOpts(); opts != nil && len(opts.JsAccDefaultDomain) > 0 {
                if defDomain, ok := opts.JsAccDefaultDomain[accName]; ok {
                        if jsEnabled {
                                s.Warnf("Skipping Default Domain %q, set for JetStream enabled account %q", defDomain, accName)
                        } else if defDomain != _EMPTY_ {
                                for src, dest := range generateJSMappingTable(defDomain) {
                                        // flip src and dest around so the domain is inserted
                                        s.Noticef("Adding default domain mapping %q -> %q to account %q %p", dest, src, accName, acc)
                                        if err := acc.AddMapping(dest, src); err != nil {
                                                s.Errorf("Error adding JetStream default domain mapping: %v", err)
                                        }
                                }
                        }
                }
        }

        s.accounts.Store(acc.Name, acc)
        s.tmpAccounts.Delete(acc.Name)
        s.enableAccountTracking(acc)

        // Can not have server lock here.
        s.mu.Unlock()
        s.registerSystemImports(acc)
        // Starting 2.9.0, we are phasing out the optimistic mode, so change
        // the account to interest-only mode (except if instructed not to do
        // it in some tests).
        if s.gateway.enabled && !gwDoNotForceInterestOnlyMode {
                s.switchAccountToInterestMode(acc.GetName())
        }
        s.mu.Lock()

        return nil
}

// Sets the account's routePoolIdx depending on presence or not of
// pooling or per-account routes. Also updates a map used by
// gateway code to retrieve a route based on some route hash.
//
// Both Server and Account lock held on entry.
func (s *Server) setRouteInfo(acc *Account) {
        // If there is a dedicated route configured for this account
        if _, ok := s.accRoutes[acc.Name]; ok {
                // We want the account name to be in the map, but we don't
                // need a value (we could store empty string)
                s.accRouteByHash.Store(acc.Name, nil)
                // Set the route pool index to -1 so that it is easy when
                // ranging over accounts to exclude those accounts when
                // trying to get accounts for a given pool index.
                acc.routePoolIdx = accDedicatedRoute
        } else {
                // If pool size more than 1, we will compute a hash code and
                // use modulo to assign to an index of the pool slice. For 1
                // and below, all accounts will be bound to the single connection
                // at index 0.
                acc.routePoolIdx = computeRoutePoolIdx(s.routesPoolSize, acc.Name)
                if s.routesPoolSize > 1 {
                        s.accRouteByHash.Store(acc.Name, acc.routePoolIdx)
                }
        }
}

// lookupAccount is a function to return the account structure
// associated with an account name.
// Lock MUST NOT be held upon entry.
func (s *Server) lookupAccount(name string) (*Account, error) {
        return s.lookupOrFetchAccount(name, true)
}

// lookupOrFetchAccount is a function to return the account structure
// associated with an account name.
// Lock MUST NOT be held upon entry.
func (s *Server) lookupOrFetchAccount(name string, fetch bool) (*Account, error) {
        var acc *Account
        if v, ok := s.accounts.Load(name); ok {
                acc = v.(*Account)
        }
        if acc != nil {
                // If we are expired and we have a resolver, then
                // return the latest information from the resolver.
                if acc.IsExpired() {
                        s.Debugf("Requested account [%s] has expired", name)
                        if s.AccountResolver() != nil && fetch {
                                if err := s.updateAccount(acc); err != nil {
                                        // This error could mask expired, so just return expired here.
                                        return nil, ErrAccountExpired
                                }
                        } else {
                                return nil, ErrAccountExpired
                        }
                }
                return acc, nil
        }
        // If we have a resolver see if it can fetch the account.
        if s.AccountResolver() == nil || !fetch {
                return nil, ErrMissingAccount
        }
        return s.fetchAccount(name)
}

// LookupAccount is a public function to return the account structure
// associated with name.
func (s *Server) LookupAccount(name string) (*Account, error) {
        return s.lookupAccount(name)
}

// This will fetch new claims and if found update the account with new claims.
// Lock MUST NOT be held upon entry.
func (s *Server) updateAccount(acc *Account) error {
        acc.mu.RLock()
        // TODO(dlc) - Make configurable
        if !acc.incomplete && time.Since(acc.updated) < time.Second {
                acc.mu.RUnlock()
                s.Debugf("Requested account update for [%s] ignored, too soon", acc.Name)
                return ErrAccountResolverUpdateTooSoon
        }
        acc.mu.RUnlock()
        claimJWT, err := s.fetchRawAccountClaims(acc.Name)
        if err != nil {
                return err
        }
        return s.updateAccountWithClaimJWT(acc, claimJWT)
}

// updateAccountWithClaimJWT will check and apply the claim update.
// Lock MUST NOT be held upon entry.
func (s *Server) updateAccountWithClaimJWT(acc *Account, claimJWT string) error {
        if acc == nil {
                return ErrMissingAccount
        }
        acc.mu.RLock()
        sameClaim := acc.claimJWT != _EMPTY_ && acc.claimJWT == claimJWT && !acc.incomplete
        acc.mu.RUnlock()
        if sameClaim {
                s.Debugf("Requested account update for [%s], same claims detected", acc.Name)
                return nil
        }
        accClaims, _, err := s.verifyAccountClaims(claimJWT)
        if err == nil && accClaims != nil {
                acc.mu.Lock()
                // if an account is updated with a different operator signing key, we want to
                // show a consistent issuer.
                acc.Issuer = accClaims.Issuer
                if acc.Name != accClaims.Subject {
                        acc.mu.Unlock()
                        return ErrAccountValidation
                }
                acc.mu.Unlock()
                s.UpdateAccountClaims(acc, accClaims)
                acc.mu.Lock()
                // needs to be set after update completed.
                // This causes concurrent calls to return with sameClaim=true if the change is effective.
                acc.claimJWT = claimJWT
                acc.mu.Unlock()
                return nil
        }
        return err
}

// fetchRawAccountClaims will grab raw account claims iff we have a resolver.
// Lock is NOT held upon entry.
func (s *Server) fetchRawAccountClaims(name string) (string, error) {
        accResolver := s.AccountResolver()
        if accResolver == nil {
                return _EMPTY_, ErrNoAccountResolver
        }
        // Need to do actual Fetch
        start := time.Now()
        claimJWT, err := fetchAccount(accResolver, name)
        fetchTime := time.Since(start)
        if fetchTime > time.Second {
                s.Warnf("Account [%s] fetch took %v", name, fetchTime)
        } else {
                s.Debugf("Account [%s] fetch took %v", name, fetchTime)
        }
        if err != nil {
                s.Warnf("Account fetch failed: %v", err)
                return "", err
        }
        return claimJWT, nil
}

// fetchAccountClaims will attempt to fetch new claims if a resolver is present.
// Lock is NOT held upon entry.
func (s *Server) fetchAccountClaims(name string) (*jwt.AccountClaims, string, error) {
        claimJWT, err := s.fetchRawAccountClaims(name)
        if err != nil {
                return nil, _EMPTY_, err
        }
        var claim *jwt.AccountClaims
        claim, claimJWT, err = s.verifyAccountClaims(claimJWT)
        if claim != nil && claim.Subject != name {
                return nil, _EMPTY_, ErrAccountValidation
        }
        return claim, claimJWT, err
}

// verifyAccountClaims will decode and validate any account claims.
func (s *Server) verifyAccountClaims(claimJWT string) (*jwt.AccountClaims, string, error) {
        accClaims, err := jwt.DecodeAccountClaims(claimJWT)
        if err != nil {
                return nil, _EMPTY_, err
        }
        if !s.isTrustedIssuer(accClaims.Issuer) {
                return nil, _EMPTY_, ErrAccountValidation
        }
        vr := jwt.CreateValidationResults()
        accClaims.Validate(vr)
        if vr.IsBlocking(true) {
                return nil, _EMPTY_, ErrAccountValidation
        }
        return accClaims, claimJWT, nil
}

// This will fetch an account from a resolver if defined.
// Lock is NOT held upon entry.
func (s *Server) fetchAccount(name string) (*Account, error) {
        accClaims, claimJWT, err := s.fetchAccountClaims(name)
        if accClaims == nil {
                return nil, err
        }
        acc := s.buildInternalAccount(accClaims)
        // Due to possible race, if registerAccount() returns a non
        // nil account, it means the same account was already
        // registered and we should use this one.
        if racc := s.registerAccount(acc); racc != nil {
                // Update with the new claims in case they are new.
                if err = s.updateAccountWithClaimJWT(racc, claimJWT); err != nil {
                        return nil, err
                }
                return racc, nil
        }
        // The sub imports may have been setup but will not have had their
        // subscriptions properly setup. Do that here.
        var needImportSubs bool

        acc.mu.Lock()
        acc.claimJWT = claimJWT
        if len(acc.imports.services) > 0 {
                if acc.ic == nil {
                        acc.ic = s.createInternalAccountClient()
                        acc.ic.acc = acc
                }
                needImportSubs = true
        }
        acc.mu.Unlock()

        // Do these outside the lock.
        if needImportSubs {
                acc.addAllServiceImportSubs()
        }

        return acc, nil
}

// Start up the server, this will not block.
//
// WaitForShutdown can be used to block and wait for the server to shutdown properly if needed
// after calling s.Shutdown()
func (s *Server) Start() {
        s.Noticef("Starting nats-server")

        gc := gitCommit
        if gc == _EMPTY_ {
                gc = "not set"
        }

        // Snapshot server options.
        opts := s.getOpts()

        // Capture if this server is a leaf that has no cluster, so we don't
        // display the cluster name if that is the case.
        s.mu.RLock()
        leafNoCluster := s.leafNoCluster
        s.mu.RUnlock()

        var clusterName string
        if !leafNoCluster {
                clusterName = s.ClusterName()
        }

        s.Noticef("  Version:  %s", VERSION)
        s.Noticef("  Git:      [%s]", gc)
        s.Debugf("  Go build: %s", s.info.GoVersion)
        if clusterName != _EMPTY_ {
                s.Noticef("  Cluster:  %s", clusterName)
        }
        s.Noticef("  Name:     %s", s.info.Name)
        if opts.JetStream {
                s.Noticef("  Node:     %s", getHash(s.info.Name))
        }
        s.Noticef("  ID:       %s", s.info.ID)

        defer s.Noticef("Server is ready")

        // Check for insecure configurations.
        s.checkAuthforWarnings()

        // Avoid RACE between Start() and Shutdown()
        s.running.Store(true)
        s.mu.Lock()
        // Update leafNodeEnabled in case options have changed post NewServer()
        // and before Start() (we should not be able to allow that, but server has
        // direct reference to user-provided options - at least before a Reload() is
        // performed.
        s.leafNodeEnabled = opts.LeafNode.Port != 0 || len(opts.LeafNode.Remotes) > 0
        s.mu.Unlock()

        s.grMu.Lock()
        s.grRunning = true
        s.grMu.Unlock()

        s.startRateLimitLogExpiration()

        // Pprof http endpoint for the profiler.
        if opts.ProfPort != 0 {
                s.StartProfiler()
        } else {
                // It's still possible to access this profile via a SYS endpoint, so set
                // this anyway. (Otherwise StartProfiler would have called it.)
                s.setBlockProfileRate(opts.ProfBlockRate)
        }

        if opts.ConfigFile != _EMPTY_ {
                var cd string
                if opts.configDigest != "" {
                        cd = fmt.Sprintf("(%s)", opts.configDigest)
                }
                s.Noticef("Using configuration file: %s %s", opts.ConfigFile, cd)
        }

        hasOperators := len(opts.TrustedOperators) > 0
        if hasOperators {
                s.Noticef("Trusted Operators")
        }
        for _, opc := range opts.TrustedOperators {
                s.Noticef("  System  : %q", opc.Audience)
                s.Noticef("  Operator: %q", opc.Name)
                s.Noticef("  Issued  : %v", time.Unix(opc.IssuedAt, 0))
                switch opc.Expires {
                case 0:
                        s.Noticef("  Expires : Never")
                default:
                        s.Noticef("  Expires : %v", time.Unix(opc.Expires, 0))
                }
        }
        if hasOperators && opts.SystemAccount == _EMPTY_ {
                s.Warnf("Trusted Operators should utilize a System Account")
        }
        if opts.MaxPayload > MAX_PAYLOAD_MAX_SIZE {
                s.Warnf("Maximum payloads over %v are generally discouraged and could lead to poor performance",
                        friendlyBytes(int64(MAX_PAYLOAD_MAX_SIZE)))
        }

        if len(opts.JsAccDefaultDomain) > 0 {
                s.Warnf("The option `default_js_domain` is a temporary backwards compatibility measure and will be removed")
        }

        // If we have a memory resolver, check the accounts here for validation exceptions.
        // This allows them to be logged right away vs when they are accessed via a client.
        if hasOperators && len(opts.resolverPreloads) > 0 {
                s.checkResolvePreloads()
        }

        // Log the pid to a file.
        if opts.PidFile != _EMPTY_ {
                if err := s.logPid(); err != nil {
                        s.Fatalf("Could not write pidfile: %v", err)
                        return
                }
        }

        // Setup system account which will start the eventing stack.
        if sa := opts.SystemAccount; sa != _EMPTY_ {
                if err := s.SetSystemAccount(sa); err != nil {
                        s.Fatalf("Can't set system account: %v", err)
                        return
                }
        } else if !opts.NoSystemAccount {
                // We will create a default system account here.
                s.SetDefaultSystemAccount()
        }

        // Start monitoring before enabling other subsystems of the
        // server to be able to monitor during startup.
        if err := s.StartMonitoring(); err != nil {
                s.Fatalf("Can't start monitoring: %v", err)
                return
        }

        // Start up resolver machinery.
        if ar := s.AccountResolver(); ar != nil {
                if err := ar.Start(s); err != nil {
                        s.Fatalf("Could not start resolver: %v", err)
                        return
                }
                // In operator mode, when the account resolver depends on an external system and
                // the system account is the bootstrapping account, start fetching it.
                if len(opts.TrustedOperators) == 1 && opts.SystemAccount != _EMPTY_ && opts.SystemAccount != DEFAULT_SYSTEM_ACCOUNT {
                        opts := s.getOpts()
                        _, isMemResolver := ar.(*MemAccResolver)
                        if v, ok := s.accounts.Load(opts.SystemAccount); !isMemResolver && ok && v.(*Account).claimJWT == _EMPTY_ {
                                s.Noticef("Using bootstrapping system account")
                                s.startGoRoutine(func() {
                                        defer s.grWG.Done()
                                        t := time.NewTicker(time.Second)
                                        defer t.Stop()
                                        for {
                                                select {
                                                case <-s.quitCh:
                                                        return
                                                case <-t.C:
                                                        sacc := s.SystemAccount()
                                                        if claimJWT, err := fetchAccount(ar, opts.SystemAccount); err != nil {
                                                                continue
                                                        } else if err = s.updateAccountWithClaimJWT(sacc, claimJWT); err != nil {
                                                                continue
                                                        }
                                                        s.Noticef("System account fetched and updated")
                                                        return
                                                }
                                        }
                                })
                        }
                }
        }

        // Start expiration of mapped GW replies, regardless if
        // this server is configured with gateway or not.
        s.startGWReplyMapExpiration()

        // Check if JetStream has been enabled. This needs to be after
        // the system account setup above. JetStream will create its
        // own system account if one is not present.
        if opts.JetStream {
                // Make sure someone is not trying to enable on the system account.
                if sa := s.SystemAccount(); sa != nil && len(sa.jsLimits) > 0 {
                        s.Fatalf("Not allowed to enable JetStream on the system account")
                }
                cfg := &JetStreamConfig{
                        StoreDir:     opts.StoreDir,
                        SyncInterval: opts.SyncInterval,
                        SyncAlways:   opts.SyncAlways,
                        Strict:       !opts.NoJetStreamStrict,
                        MaxMemory:    opts.JetStreamMaxMemory,
                        MaxStore:     opts.JetStreamMaxStore,
                        Domain:       opts.JetStreamDomain,
                        CompressOK:   true,
                        UniqueTag:    opts.JetStreamUniqueTag,
                }
                if err := s.EnableJetStream(cfg); err != nil {
                        s.Fatalf("Can't start JetStream: %v", err)
                        return
                }
        } else {
                // Check to see if any configured accounts have JetStream enabled.
                sa, ga := s.SystemAccount(), s.GlobalAccount()
                var hasSys, hasGlobal bool
                var total int

                s.accounts.Range(func(k, v any) bool {
                        total++
                        acc := v.(*Account)
                        if acc == sa {
                                hasSys = true
                        } else if acc == ga {
                                hasGlobal = true
                        }
                        acc.mu.RLock()
                        hasJs := len(acc.jsLimits) > 0
                        acc.mu.RUnlock()
                        if hasJs {
                                s.checkJetStreamExports()
                                acc.enableAllJetStreamServiceImportsAndMappings()
                        }
                        return true
                })
                // If we only have the system account and the global account and we are not standalone,
                // go ahead and enable JS on $G in case we are in simple mixed mode setup.
                if total == 2 && hasSys && hasGlobal && !s.standAloneMode() {
                        ga.mu.Lock()
                        ga.jsLimits = map[string]JetStreamAccountLimits{
                                _EMPTY_: dynamicJSAccountLimits,
                        }
                        ga.mu.Unlock()
                        s.checkJetStreamExports()
                        ga.enableAllJetStreamServiceImportsAndMappings()
                }
        }

        // Delayed API response handling. Start regardless of JetStream being
        // currently configured or not (since it can be enabled/disabled with
        // configuration reload).
        s.startGoRoutine(s.delayedAPIResponder)

        // Start OCSP Stapling monitoring for TLS certificates if enabled. Hook TLS handshake for
        // OCSP check on peers (LEAF and CLIENT kind) if enabled.
        s.startOCSPMonitoring()

        // Configure OCSP Response Cache for peer OCSP checks if enabled.
        s.initOCSPResponseCache()

        // Start up gateway if needed. Do this before starting the routes, because
        // we want to resolve the gateway host:port so that this information can
        // be sent to other routes.
        if opts.Gateway.Port != 0 {
                s.startGateways()
        }

        // Start websocket server if needed. Do this before starting the routes, and
        // leaf node because we want to resolve the gateway host:port so that this
        // information can be sent to other routes.
        if opts.Websocket.Port != 0 {
                s.startWebsocketServer()
        }

        // Start up listen if we want to accept leaf node connections.
        if opts.LeafNode.Port != 0 {
                // Will resolve or assign the advertise address for the leafnode listener.
                // We need that in StartRouting().
                s.startLeafNodeAcceptLoop()
        }

        // Solicit remote servers for leaf node connections.
        if len(opts.LeafNode.Remotes) > 0 {
                s.solicitLeafNodeRemotes(opts.LeafNode.Remotes)
        }

        // TODO (ik): I wanted to refactor this by starting the client
        // accept loop first, that is, it would resolve listen spec
        // in place, but start the accept-for-loop in a different go
        // routine. This would get rid of the synchronization between
        // this function and StartRouting, which I also would have wanted
        // to refactor, but both AcceptLoop() and StartRouting() have
        // been exported and not sure if that would break users using them.
        // We could mark them as deprecated and remove in a release or two...

        // The Routing routine needs to wait for the client listen
        // port to be opened and potential ephemeral port selected.
        clientListenReady := make(chan struct{})

        // MQTT
        if opts.MQTT.Port != 0 {
                s.startMQTT()
        }

        // Start up routing as well if needed.
        if opts.Cluster.Port != 0 {
                s.startGoRoutine(func() {
                        s.StartRouting(clientListenReady)
                })
        }

        if opts.PortsFileDir != _EMPTY_ {
                s.logPorts()
        }

        if opts.TLSRateLimit > 0 {
                s.startGoRoutine(s.logRejectedTLSConns)
        }

        // We've finished starting up.
        close(s.startupComplete)

        // Wait for clients.
        if !opts.DontListen {
                s.AcceptLoop(clientListenReady)
        }

        // Bring OSCP Response cache online after accept loop started in anticipation of NATS-enabled cache types
        s.startOCSPResponseCache()
}

func (s *Server) isShuttingDown() bool {
        return s.shutdown.Load()
}

// Shutdown will shutdown the server instance by kicking out the AcceptLoop
// and closing all associated clients.
func (s *Server) Shutdown() {
        if s == nil {
                return
        }
        // This is for JetStream R1 Pull Consumers to allow signaling
        // that pending pull requests are invalid.
        s.signalPullConsumers()

        // Transfer off any raft nodes that we are a leader by stepping them down.
        s.stepdownRaftNodes()

        // Shutdown the eventing system as needed.
        // This is done first to send out any messages for
        // account status. We will also clean up any
        // eventing items associated with accounts.
        s.shutdownEventing()

        // Prevent issues with multiple calls.
        if s.isShuttingDown() {
                return
        }

        s.mu.Lock()
        s.Noticef("Initiating Shutdown...")

        accRes := s.accResolver

        opts := s.getOpts()

        s.shutdown.Store(true)
        s.running.Store(false)
        s.grMu.Lock()
        s.grRunning = false
        s.grMu.Unlock()
        s.mu.Unlock()

        if accRes != nil {
                accRes.Close()
        }

        // Now check and shutdown jetstream.
        s.shutdownJetStream()

        // Now shutdown the nodes
        s.shutdownRaftNodes()

        s.mu.Lock()
        conns := make(map[uint64]*client)

        // Copy off the clients
        for i, c := range s.clients {
                conns[i] = c
        }
        // Copy off the connections that are not yet registered
        // in s.routes, but for which the readLoop has started
        s.grMu.Lock()
        for i, c := range s.grTmpClients {
                conns[i] = c
        }
        s.grMu.Unlock()
        // Copy off the routes
        s.forEachRoute(func(r *client) {
                r.mu.Lock()
                conns[r.cid] = r
                r.mu.Unlock()
        })
        // Copy off the gateways
        s.getAllGatewayConnections(conns)

        // Copy off the leaf nodes
        for i, c := range s.leafs {
                conns[i] = c
        }

        // Number of done channel responses we expect.
        doneExpected := 0

        // Kick client AcceptLoop()
        if s.listener != nil {
                doneExpected++
                s.listener.Close()
                s.listener = nil
        }

        // Kick websocket server
        doneExpected += s.closeWebsocketServer()

        // Kick MQTT accept loop
        if s.mqtt.listener != nil {
                doneExpected++
                s.mqtt.listener.Close()
                s.mqtt.listener = nil
        }

        // Kick leafnodes AcceptLoop()
        if s.leafNodeListener != nil {
                doneExpected++
                s.leafNodeListener.Close()
                s.leafNodeListener = nil
        }

        // Kick route AcceptLoop()
        if s.routeListener != nil {
                doneExpected++
                s.routeListener.Close()
                s.routeListener = nil
        }

        // Kick Gateway AcceptLoop()
        if s.gatewayListener != nil {
                doneExpected++
                s.gatewayListener.Close()
                s.gatewayListener = nil
        }

        // Kick HTTP monitoring if its running
        if s.http != nil {
                doneExpected++
                s.http.Close()
                s.http = nil
        }

        // Kick Profiling if its running
        if s.profiler != nil {
                doneExpected++
                s.profiler.Close()
        }

        s.mu.Unlock()

        // Release go routines that wait on that channel
        close(s.quitCh)

        // Close client and route connections
        for _, c := range conns {
                c.setNoReconnect()
                c.closeConnection(ServerShutdown)
        }

        // Block until the accept loops exit
        for doneExpected > 0 {
                <-s.done
                doneExpected--
        }

        // Wait for go routines to be done.
        s.grWG.Wait()

        if opts.PortsFileDir != _EMPTY_ {
                s.deletePortsFile(opts.PortsFileDir)
        }

        s.Noticef("Server Exiting..")

        // Stop OCSP Response Cache
        if s.ocsprc != nil {
                s.ocsprc.Stop(s)
        }

        // Close logger if applicable. It allows tests on Windows
        // to be able to do proper cleanup (delete log file).
        s.logging.RLock()
        log := s.logging.logger
        s.logging.RUnlock()
        if log != nil {
                if l, ok := log.(*logger.Logger); ok {
                        l.Close()
                }
        }
        // Notify that the shutdown is complete
        close(s.shutdownComplete)
}

// Close the websocket server if running. If so, returns 1, else 0.
// Server lock held on entry.
func (s *Server) closeWebsocketServer() int {
        ws := &s.websocket
        ws.mu.Lock()
        hs := ws.server
        if hs != nil {
                ws.server = nil
                ws.listener = nil
        }
        ws.mu.Unlock()
        if hs != nil {
                hs.Close()
                return 1
        }
        return 0
}

// WaitForShutdown will block until the server has been fully shutdown.
func (s *Server) WaitForShutdown() {
        <-s.shutdownComplete
}

// AcceptLoop is exported for easier testing.
func (s *Server) AcceptLoop(clr chan struct{}) {
        // If we were to exit before the listener is setup properly,
        // make sure we close the channel.
        defer func() {
                if clr != nil {
                        close(clr)
                }
        }()

        if s.isShuttingDown() {
                return
        }

        // Snapshot server options.
        opts := s.getOpts()

        // Setup state that can enable shutdown
        s.mu.Lock()
        hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
        l, e := s.getServerListener(hp)
        s.listenerErr = e
        if e != nil {
                s.mu.Unlock()
                s.Fatalf("Error listening on port: %s, %q", hp, e)
                return
        }
        s.Noticef("Listening for client connections on %s",
                net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))

        // Alert of TLS enabled.
        if opts.TLSConfig != nil {
                s.Noticef("TLS required for client connections")
                if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
                        s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
                }
        }

        // If server was started with RANDOM_PORT (-1), opts.Port would be equal
        // to 0 at the beginning this function. So we need to get the actual port
        if opts.Port == 0 {
                // Write resolved port back to options.
                opts.Port = l.Addr().(*net.TCPAddr).Port
        }

        // Now that port has been set (if it was set to RANDOM), set the
        // server's info Host/Port with either values from Options or
        // ClientAdvertise.
        if err := s.setInfoHostPort(); err != nil {
                s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err)
                l.Close()
                s.mu.Unlock()
                return
        }
        // Keep track of client connect URLs. We may need them later.
        s.clientConnectURLs = s.getClientConnectURLs()
        s.listener = l

        go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
                func(_ error) bool {
                        if s.isLameDuckMode() {
                                // Signal that we are not accepting new clients
                                s.ldmCh <- true
                                // Now wait for the Shutdown...
                                <-s.quitCh
                                return true
                        }
                        return false
                })
        s.mu.Unlock()

        // Let the caller know that we are ready
        close(clr)
        clr = nil
}

// getServerListener returns a network listener for the given host-port address.
// If the Server already has an active listener (s.listener), it returns that listener
// along with any previous error (s.listenerErr). Otherwise, it creates and returns
// a new TCP listener on the specified address using natsListen.
func (s *Server) getServerListener(hp string) (net.Listener, error) {
        if s.listener != nil {
                return s.listener, s.listenerErr
        }

        return natsListen("tcp", hp)
}

// InProcessConn returns an in-process connection to the server,
// avoiding the need to use a TCP listener for local connectivity
// within the same process. This can be used regardless of the
// state of the DontListen option.
func (s *Server) InProcessConn() (net.Conn, error) {
        pl, pr := net.Pipe()
        if !s.startGoRoutine(func() {
                s.createClientInProcess(pl)
                s.grWG.Done()
        }) {
                pl.Close()
                pr.Close()
                return nil, fmt.Errorf("failed to create connection")
        }
        return pr, nil
}

func (s *Server) acceptConnections(l net.Listener, acceptName string, createFunc func(conn net.Conn), errFunc func(err error) bool) {
        tmpDelay := ACCEPT_MIN_SLEEP

        for {
                conn, err := l.Accept()
                if err != nil {
                        if errFunc != nil && errFunc(err) {
                                return
                        }
                        if tmpDelay = s.acceptError(acceptName, err, tmpDelay); tmpDelay < 0 {
                                break
                        }
                        continue
                }
                tmpDelay = ACCEPT_MIN_SLEEP
                if !s.startGoRoutine(func() {
                        s.reloadMu.RLock()
                        createFunc(conn)
                        s.reloadMu.RUnlock()
                        s.grWG.Done()
                }) {
                        conn.Close()
                }
        }
        s.Debugf(acceptName + " accept loop exiting..")
        s.done <- true
}

// This function sets the server's info Host/Port based on server Options.
// Note that this function may be called during config reload, this is why
// Host/Port may be reset to original Options if the ClientAdvertise option
// is not set (since it may have previously been).
func (s *Server) setInfoHostPort() error {
        // When this function is called, opts.Port is set to the actual listen
        // port (if option was originally set to RANDOM), even during a config
        // reload. So use of s.opts.Port is safe.
        opts := s.getOpts()
        if opts.ClientAdvertise != _EMPTY_ {
                h, p, err := parseHostPort(opts.ClientAdvertise, opts.Port)
                if err != nil {
                        return err
                }
                s.info.Host = h
                s.info.Port = p
        } else {
                s.info.Host = opts.Host
                s.info.Port = opts.Port
        }
        return nil
}

// StartProfiler is called to enable dynamic profiling.
func (s *Server) StartProfiler() {
        if s.isShuttingDown() {
                return
        }

        // Snapshot server options.
        opts := s.getOpts()

        port := opts.ProfPort

        // Check for Random Port
        if port == -1 {
                port = 0
        }

        s.mu.Lock()
        hp := net.JoinHostPort(opts.Host, strconv.Itoa(port))
        l, err := net.Listen("tcp", hp)

        if err != nil {
                s.mu.Unlock()
                s.Fatalf("error starting profiler: %s", err)
                return
        }
        s.Noticef("profiling port: %d", l.Addr().(*net.TCPAddr).Port)

        srv := &http.Server{
                Addr:           hp,
                Handler:        http.DefaultServeMux,
                MaxHeaderBytes: 1 << 20,
                ReadTimeout:    time.Second * 5,
        }
        s.profiler = l
        s.profilingServer = srv

        s.setBlockProfileRate(opts.ProfBlockRate)

        go func() {
                // if this errors out, it's probably because the server is being shutdown
                err := srv.Serve(l)
                if err != nil {
                        if !s.isShuttingDown() {
                                s.Fatalf("error starting profiler: %s", err)
                        }
                }
                srv.Close()
                s.done <- true
        }()
        s.mu.Unlock()
}

func (s *Server) setBlockProfileRate(rate int) {
        // Passing i ProfBlockRate <= 0 here will disable or > 0 will enable.
        runtime.SetBlockProfileRate(rate)

        if rate > 0 {
                s.Warnf("Block profiling is enabled (rate %d), this may have a performance impact", rate)
        }
}

// StartHTTPMonitoring will enable the HTTP monitoring port.
// DEPRECATED: Should use StartMonitoring.
func (s *Server) StartHTTPMonitoring() {
        s.startMonitoring(false)
}

// StartHTTPSMonitoring will enable the HTTPS monitoring port.
// DEPRECATED: Should use StartMonitoring.
func (s *Server) StartHTTPSMonitoring() {
        s.startMonitoring(true)
}

// StartMonitoring starts the HTTP or HTTPs server if needed.
func (s *Server) StartMonitoring() error {
        // Snapshot server options.
        opts := s.getOpts()

        // Specifying both HTTP and HTTPS ports is a misconfiguration
        if opts.HTTPPort != 0 && opts.HTTPSPort != 0 {
                return fmt.Errorf("can't specify both HTTP (%v) and HTTPs (%v) ports", opts.HTTPPort, opts.HTTPSPort)
        }
        var err error
        if opts.HTTPPort != 0 {
                err = s.startMonitoring(false)
        } else if opts.HTTPSPort != 0 {
                if opts.TLSConfig == nil {
                        return fmt.Errorf("TLS cert and key required for HTTPS")
                }
                err = s.startMonitoring(true)
        }
        return err
}

// HTTP endpoints
const (
        RootPath         = "/"
        VarzPath         = "/varz"
        ConnzPath        = "/connz"
        RoutezPath       = "/routez"
        GatewayzPath     = "/gatewayz"
        LeafzPath        = "/leafz"
        SubszPath        = "/subsz"
        StackszPath      = "/stacksz"
        AccountzPath     = "/accountz"
        AccountStatzPath = "/accstatz"
        JszPath          = "/jsz"
        HealthzPath      = "/healthz"
        IPQueuesPath     = "/ipqueuesz"
        RaftzPath        = "/raftz"
        ExpvarzPath      = "/debug/vars"
)

func (s *Server) basePath(p string) string {
        return path.Join(s.httpBasePath, p)
}

type captureHTTPServerLog struct {
        s      *Server
        prefix string
}

func (cl *captureHTTPServerLog) Write(p []byte) (int, error) {
        var buf [128]byte
        var b = buf[:0]

        b = append(b, []byte(cl.prefix)...)
        offset := 0
        if bytes.HasPrefix(p, []byte("http:")) {
                offset = 6
        }
        b = append(b, p[offset:]...)
        cl.s.Errorf(string(b))
        return len(p), nil
}

// The TLS configuration is passed to the listener when the monitoring
// "server" is setup. That prevents TLS configuration updates on reload
// from being used. By setting this function in tls.Config.GetConfigForClient
// we instruct the TLS handshake to ask for the tls configuration to be
// used for a specific client. We don't care which client, we always use
// the same TLS configuration.
func (s *Server) getMonitoringTLSConfig(_ *tls.ClientHelloInfo) (*tls.Config, error) {
        opts := s.getOpts()
        tc := opts.TLSConfig.Clone()
        tc.ClientAuth = tls.NoClientCert
        return tc, nil
}

// Start the monitoring server
func (s *Server) startMonitoring(secure bool) error {
        if s.isShuttingDown() {
                return nil
        }

        // Snapshot server options.
        opts := s.getOpts()

        var (
                hp           string
                err          error
                httpListener net.Listener
                port         int
        )

        monitorProtocol := "http"

        if secure {
                monitorProtocol += "s"
                port = opts.HTTPSPort
                if port == -1 {
                        port = 0
                }
                hp = net.JoinHostPort(opts.HTTPHost, strconv.Itoa(port))
                config := opts.TLSConfig.Clone()
                if !s.ocspPeerVerify {
                        config.GetConfigForClient = s.getMonitoringTLSConfig
                        config.ClientAuth = tls.NoClientCert
                }
                httpListener, err = tls.Listen("tcp", hp, config)

        } else {
                port = opts.HTTPPort
                if port == -1 {
                        port = 0
                }
                hp = net.JoinHostPort(opts.HTTPHost, strconv.Itoa(port))
                httpListener, err = net.Listen("tcp", hp)
        }

        if err != nil {
                return fmt.Errorf("can't listen to the monitor port: %v", err)
        }

        rport := httpListener.Addr().(*net.TCPAddr).Port
        s.Noticef("Starting %s monitor on %s", monitorProtocol, net.JoinHostPort(opts.HTTPHost, strconv.Itoa(rport)))

        mux := http.NewServeMux()

        // Root
        mux.HandleFunc(s.basePath(RootPath), s.HandleRoot)
        // Varz
        mux.HandleFunc(s.basePath(VarzPath), s.HandleVarz)
        // Connz
        mux.HandleFunc(s.basePath(ConnzPath), s.HandleConnz)
        // Routez
        mux.HandleFunc(s.basePath(RoutezPath), s.HandleRoutez)
        // Gatewayz
        mux.HandleFunc(s.basePath(GatewayzPath), s.HandleGatewayz)
        // Leafz
        mux.HandleFunc(s.basePath(LeafzPath), s.HandleLeafz)
        // Subz
        mux.HandleFunc(s.basePath(SubszPath), s.HandleSubsz)
        // Subz alias for backwards compatibility
        mux.HandleFunc(s.basePath("/subscriptionsz"), s.HandleSubsz)
        // Stacksz
        mux.HandleFunc(s.basePath(StackszPath), s.HandleStacksz)
        // Accountz
        mux.HandleFunc(s.basePath(AccountzPath), s.HandleAccountz)
        // Accstatz
        mux.HandleFunc(s.basePath(AccountStatzPath), s.HandleAccountStatz)
        // Jsz
        mux.HandleFunc(s.basePath(JszPath), s.HandleJsz)
        // Healthz
        mux.HandleFunc(s.basePath(HealthzPath), s.HandleHealthz)
        // IPQueuesz
        mux.HandleFunc(s.basePath(IPQueuesPath), s.HandleIPQueuesz)
        // Raftz
        mux.HandleFunc(s.basePath(RaftzPath), s.HandleRaftz)
        // Expvarz
        mux.Handle(s.basePath(ExpvarzPath), expvar.Handler())

        // Do not set a WriteTimeout because it could cause cURL/browser
        // to return empty response or unable to display page if the
        // server needs more time to build the response.
        srv := &http.Server{
                Addr:              hp,
                Handler:           mux,
                MaxHeaderBytes:    1 << 20,
                ErrorLog:          log.New(&captureHTTPServerLog{s, "monitoring: "}, _EMPTY_, 0),
                ReadHeaderTimeout: time.Second * 5,
        }
        s.mu.Lock()
        s.http = httpListener
        s.httpHandler = mux
        s.monitoringServer = srv
        s.mu.Unlock()

        go func() {
                if err := srv.Serve(httpListener); err != nil {
                        if !s.isShuttingDown() {
                                s.Fatalf("Error starting monitor on %q: %v", hp, err)
                        }
                }
                srv.Close()
                s.mu.Lock()
                s.httpHandler = nil
                s.mu.Unlock()
                s.done <- true
        }()

        return nil
}

// HTTPHandler returns the http.Handler object used to handle monitoring
// endpoints. It will return nil if the server is not configured for
// monitoring, or if the server has not been started yet (Server.Start()).
func (s *Server) HTTPHandler() http.Handler {
        s.mu.Lock()
        defer s.mu.Unlock()
        return s.httpHandler
}

// Perform a conditional deep copy due to reference nature of [Client|WS]ConnectURLs.
// If updates are made to Info, this function should be consulted and updated.
// Assume lock is held.
func (s *Server) copyInfo() Info {
        info := s.info
        if len(info.ClientConnectURLs) > 0 {
                info.ClientConnectURLs = append([]string(nil), s.info.ClientConnectURLs...)
        }
        if len(info.WSConnectURLs) > 0 {
                info.WSConnectURLs = append([]string(nil), s.info.WSConnectURLs...)
        }
        return info
}

// tlsMixConn is used when we can receive both TLS and non-TLS connections on same port.
type tlsMixConn struct {
        net.Conn
        pre *bytes.Buffer
}

// Read for our mixed multi-reader.
func (c *tlsMixConn) Read(b []byte) (int, error) {
        if c.pre != nil {
                n, err := c.pre.Read(b)
                if c.pre.Len() == 0 {
                        c.pre = nil
                }
                return n, err
        }
        return c.Conn.Read(b)
}

func (s *Server) createClient(conn net.Conn) *client {
        return s.createClientEx(conn, false)
}

func (s *Server) createClientInProcess(conn net.Conn) *client {
        return s.createClientEx(conn, true)
}

func (s *Server) createClientEx(conn net.Conn, inProcess bool) *client {
        // Snapshot server options.
        opts := s.getOpts()

        maxPay := int32(opts.MaxPayload)
        maxSubs := int32(opts.MaxSubs)
        // For system, maxSubs of 0 means unlimited, so re-adjust here.
        if maxSubs == 0 {
                maxSubs = -1
        }
        now := time.Now()

        c := &client{
                srv:   s,
                nc:    conn,
                opts:  defaultOpts,
                mpay:  maxPay,
                msubs: maxSubs,
                start: now,
                last:  now,
                iproc: inProcess,
        }

        c.registerWithAccount(s.globalAccount())

        var info Info
        var authRequired bool

        s.mu.Lock()
        // Grab JSON info string
        info = s.copyInfo()
        if s.nonceRequired() {
                // Nonce handling
                var raw [nonceLen]byte
                nonce := raw[:]
                s.generateNonce(nonce)
                info.Nonce = string(nonce)
        }
        c.nonce = []byte(info.Nonce)
        authRequired = info.AuthRequired

        // Check to see if we have auth_required set but we also have a no_auth_user.
        // If so set back to false.
        if info.AuthRequired && opts.NoAuthUser != _EMPTY_ && opts.NoAuthUser != s.sysAccOnlyNoAuthUser {
                info.AuthRequired = false
        }

        // Check to see if this is an in-process connection with tls_required.
        // If so, set as not required, but available.
        if inProcess && info.TLSRequired {
                info.TLSRequired = false
                info.TLSAvailable = true
        }

        s.totalClients++
        s.mu.Unlock()

        // Grab lock
        c.mu.Lock()
        if authRequired {
                c.flags.set(expectConnect)
        }

        // Initialize
        c.initClient()

        c.Debugf("Client connection created")

        // Save info.TLSRequired value since we may neeed to change it back and forth.
        orgInfoTLSReq := info.TLSRequired

        var tlsFirstFallback time.Duration
        // Check if we should do TLS first.
        tlsFirst := opts.TLSConfig != nil && opts.TLSHandshakeFirst
        if tlsFirst {
                // Make sure info.TLSRequired is set to true (it could be false
                // if AllowNonTLS is enabled).
                info.TLSRequired = true
                // Get the fallback delay value if applicable.
                if f := opts.TLSHandshakeFirstFallback; f > 0 {
                        tlsFirstFallback = f
                } else if inProcess {
                        // For in-process connection, we will always have a fallback
                        // delay. It allows support for non-TLS, TLS and "TLS First"
                        // in-process clients to successfully connect.
                        tlsFirstFallback = DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY
                }
        }

        // Decide if we are going to require TLS or not and generate INFO json.
        tlsRequired := info.TLSRequired
        infoBytes := c.generateClientInfoJSON(info)

        // Send our information, except if TLS and TLSHandshakeFirst is requested.
        if !tlsFirst {
                // Need to be sent in place since writeLoop cannot be started until
                // TLS handshake is done (if applicable).
                c.sendProtoNow(infoBytes)
        }

        // Unlock to register
        c.mu.Unlock()

        // Register with the server.
        s.mu.Lock()
        // If server is not running, Shutdown() may have already gathered the
        // list of connections to close. It won't contain this one, so we need
        // to bail out now otherwise the readLoop started down there would not
        // be interrupted. Skip also if in lame duck mode.
        if !s.isRunning() || s.ldm {
                // There are some tests that create a server but don't start it,
                // and use "async" clients and perform the parsing manually. Such
                // clients would branch here (since server is not running). However,
                // when a server was really running and has been shutdown, we must
                // close this connection.
                if s.isShuttingDown() {
                        conn.Close()
                }
                s.mu.Unlock()
                return c
        }

        // If there is a max connections specified, check that adding
        // this new client would not push us over the max
        if opts.MaxConn > 0 && len(s.clients) >= opts.MaxConn {
                s.mu.Unlock()
                c.maxConnExceeded()
                return nil
        }
        s.clients[c.cid] = c

        s.mu.Unlock()

        // Re-Grab lock
        c.mu.Lock()

        isClosed := c.isClosed()
        var pre []byte
        // We need first to check for "TLS First" fallback delay.
        if !isClosed && tlsFirstFallback > 0 {
                // We wait and see if we are getting any data. Since we did not send
                // the INFO protocol yet, only clients that use TLS first should be
                // sending data (the TLS handshake). We don't really check the content:
                // if it is a rogue agent and not an actual client performing the
                // TLS handshake, the error will be detected when performing the
                // handshake on our side.
                pre = make([]byte, 4)
                c.nc.SetReadDeadline(time.Now().Add(tlsFirstFallback))
                n, _ := io.ReadFull(c.nc, pre[:])
                c.nc.SetReadDeadline(time.Time{})
                // If we get any data (regardless of possible timeout), we will proceed
                // with the TLS handshake.
                if n > 0 {
                        pre = pre[:n]
                } else {
                        // We did not get anything so we will send the INFO protocol.
                        pre = nil

                        // Restore the original info.TLSRequired value if it is
                        // different that the current value and regenerate infoBytes.
                        if orgInfoTLSReq != info.TLSRequired {
                                info.TLSRequired = orgInfoTLSReq
                                infoBytes = c.generateClientInfoJSON(info)
                        }
                        c.sendProtoNow(infoBytes)
                        // Set the boolean to false for the rest of the function.
                        tlsFirst = false
                        // Check closed status again
                        isClosed = c.isClosed()
                }
        }
        // If we have both TLS and non-TLS allowed we need to see which
        // one the client wants. We'll always allow this for in-process
        // connections.
        if !isClosed && !tlsFirst && opts.TLSConfig != nil && (inProcess || opts.AllowNonTLS) {
                pre = make([]byte, 4)
                c.nc.SetReadDeadline(time.Now().Add(secondsToDuration(opts.TLSTimeout)))
                n, _ := io.ReadFull(c.nc, pre[:])
                c.nc.SetReadDeadline(time.Time{})
                pre = pre[:n]
                if n > 0 && pre[0] == 0x16 {
                        tlsRequired = true
                } else {
                        tlsRequired = false
                }
        }

        // Check for TLS
        if !isClosed && tlsRequired {
                if s.connRateCounter != nil && !s.connRateCounter.allow() {
                        c.mu.Unlock()
                        c.sendErr("Connection throttling is active. Please try again later.")
                        c.closeConnection(MaxConnectionsExceeded)
                        return nil
                }

                // If we have a prebuffer create a multi-reader.
                if len(pre) > 0 {
                        c.nc = &tlsMixConn{c.nc, bytes.NewBuffer(pre)}
                        // Clear pre so it is not parsed.
                        pre = nil
                }
                // Performs server-side TLS handshake.
                if err := c.doTLSServerHandshake(_EMPTY_, opts.TLSConfig, opts.TLSTimeout, opts.TLSPinnedCerts); err != nil {
                        c.mu.Unlock()
                        return nil
                }
        }

        // Now, send the INFO if it was delayed
        if !isClosed && tlsFirst {
                c.flags.set(didTLSFirst)
                c.sendProtoNow(infoBytes)
                // Check closed status
                isClosed = c.isClosed()
        }

        // Connection could have been closed while sending the INFO proto.
        if isClosed {
                c.mu.Unlock()
                // We need to call closeConnection() to make sure that proper cleanup is done.
                c.closeConnection(WriteError)
                return nil
        }

        // Check for Auth. We schedule this timer after the TLS handshake to avoid
        // the race where the timer fires during the handshake and causes the
        // server to write bad data to the socket. See issue #432.
        if authRequired {
                c.setAuthTimer(secondsToDuration(opts.AuthTimeout))
        }

        // Do final client initialization

        // Set the Ping timer. Will be reset once connect was received.
        c.setPingTimer()

        // Spin up the read loop.
        s.startGoRoutine(func() { c.readLoop(pre) })

        // Spin up the write loop.
        s.startGoRoutine(func() { c.writeLoop() })

        if tlsRequired {
                c.Debugf("TLS handshake complete")
                cs := c.nc.(*tls.Conn).ConnectionState()
                c.Debugf("TLS version %s, cipher suite %s", tlsVersion(cs.Version), tls.CipherSuiteName(cs.CipherSuite))
        }

        c.mu.Unlock()

        return c
}

// This will save off a closed client in a ring buffer such that
// /connz can inspect. Useful for debugging, etc.
func (s *Server) saveClosedClient(c *client, nc net.Conn, subs map[string]*subscription, reason ClosedState) {
        now := time.Now()

        s.accountDisconnectEvent(c, now, reason.String())

        c.mu.Lock()

        cc := &closedClient{}
        cc.fill(c, nc, now, false)
        // Note that cc.fill is using len(c.subs), which may have been set to nil by now,
        // so replace cc.NumSubs with len(subs).
        cc.NumSubs = uint32(len(subs))
        cc.Stop = &now
        cc.Reason = reason.String()

        // Do subs, do not place by default in main ConnInfo
        if len(subs) > 0 {
                cc.subs = make([]SubDetail, 0, len(subs))
                for _, sub := range subs {
                        cc.subs = append(cc.subs, newSubDetail(sub))
                }
        }
        // Hold user as well.
        cc.user = c.getRawAuthUser()
        // Hold account name if not the global account.
        if c.acc != nil && c.acc.Name != globalAccountName {
                cc.acc = c.acc.Name
        }
        cc.JWT = c.opts.JWT
        cc.IssuerKey = issuerForClient(c)
        cc.Tags = c.tags
        cc.NameTag = c.nameTag
        c.mu.Unlock()

        // Place in the ring buffer
        s.mu.Lock()
        if s.closed != nil {
                s.closed.append(cc)
        }
        s.mu.Unlock()
}

// Adds to the list of client and websocket clients connect URLs.
// If there was a change, an INFO protocol is sent to registered clients
// that support async INFO protocols.
// Server lock held on entry.
func (s *Server) addConnectURLsAndSendINFOToClients(curls, wsurls []string) {
        s.updateServerINFOAndSendINFOToClients(curls, wsurls, true)
}

// Removes from the list of client and websocket clients connect URLs.
// If there was a change, an INFO protocol is sent to registered clients
// that support async INFO protocols.
// Server lock held on entry.
func (s *Server) removeConnectURLsAndSendINFOToClients(curls, wsurls []string) {
        s.updateServerINFOAndSendINFOToClients(curls, wsurls, false)
}

// Updates the list of client and websocket clients connect URLs and if any change
// sends an async INFO update to clients that support it.
// Server lock held on entry.
func (s *Server) updateServerINFOAndSendINFOToClients(curls, wsurls []string, add bool) {
        remove := !add
        // Will return true if we need alter the server's Info object.
        updateMap := func(urls []string, m refCountedUrlSet) bool {
                wasUpdated := false
                for _, url := range urls {
                        if add && m.addUrl(url) {
                                wasUpdated = true
                        } else if remove && m.removeUrl(url) {
                                wasUpdated = true
                        }
                }
                return wasUpdated
        }
        cliUpdated := updateMap(curls, s.clientConnectURLsMap)
        wsUpdated := updateMap(wsurls, s.websocket.connectURLsMap)

        updateInfo := func(infoURLs *[]string, urls []string, m refCountedUrlSet) {
                // Recreate the info's slice from the map
                *infoURLs = (*infoURLs)[:0]
                // Add this server client connect ULRs first...
                *infoURLs = append(*infoURLs, urls...)
                // Then the ones from the map
                for url := range m {
                        *infoURLs = append(*infoURLs, url)
                }
        }
        if cliUpdated {
                updateInfo(&s.info.ClientConnectURLs, s.clientConnectURLs, s.clientConnectURLsMap)
        }
        if wsUpdated {
                updateInfo(&s.info.WSConnectURLs, s.websocket.connectURLs, s.websocket.connectURLsMap)
        }
        if cliUpdated || wsUpdated {
                // Send to all registered clients that support async INFO protocols.
                s.sendAsyncInfoToClients(cliUpdated, wsUpdated)
        }
}

// Handle closing down a connection when the handshake has timedout.
func tlsTimeout(c *client, conn *tls.Conn) {
        c.mu.Lock()
        closed := c.isClosed()
        c.mu.Unlock()
        // Check if already closed
        if closed {
                return
        }
        cs := conn.ConnectionState()
        if !cs.HandshakeComplete {
                c.Errorf("TLS handshake timeout")
                c.sendErr("Secure Connection - TLS Required")
                c.closeConnection(TLSHandshakeError)
        }
}

// Seems silly we have to write these
func tlsVersion(ver uint16) string {
        switch ver {
        case tls.VersionTLS10:
                return "1.0"
        case tls.VersionTLS11:
                return "1.1"
        case tls.VersionTLS12:
                return "1.2"
        case tls.VersionTLS13:
                return "1.3"
        }
        return fmt.Sprintf("Unknown [0x%x]", ver)
}

func tlsVersionFromString(ver string) (uint16, error) {
        switch ver {
        case "1.0":
                return tls.VersionTLS10, nil
        case "1.1":
                return tls.VersionTLS11, nil
        case "1.2":
                return tls.VersionTLS12, nil
        case "1.3":
                return tls.VersionTLS13, nil
        }
        return 0, fmt.Errorf("unknown version: %v", ver)
}

// Remove a client or route from our internal accounting.
func (s *Server) removeClient(c *client) {
        // kind is immutable, so can check without lock
        switch c.kind {
        case CLIENT:
                c.mu.Lock()
                cid := c.cid
                updateProtoInfoCount := false
                if c.kind == CLIENT && c.opts.Protocol >= ClientProtoInfo {
                        updateProtoInfoCount = true
                }
                proxyKey := c.proxyKey
                c.mu.Unlock()

                s.mu.Lock()
                delete(s.clients, cid)
                if updateProtoInfoCount {
                        s.cproto--
                }
                if proxyKey != _EMPTY_ {
                        s.removeProxiedConn(proxyKey, cid)
                }
                s.mu.Unlock()
        case ROUTER:
                s.removeRoute(c)
        case GATEWAY:
                s.removeRemoteGatewayConnection(c)
        case LEAF:
                s.removeLeafNodeConnection(c)
        }
}

// Remove the connection with id `cid` from the map of connections
// under the public key `key` of the trusted proxies.
//
// Server lock must be held on entry.
func (s *Server) removeProxiedConn(key string, cid uint64) {
        conns := s.proxiedConns[key]
        delete(conns, cid)
        if len(conns) == 0 {
                delete(s.proxiedConns, key)
        }
}

func (s *Server) removeFromTempClients(cid uint64) {
        s.grMu.Lock()
        delete(s.grTmpClients, cid)
        s.grMu.Unlock()
}

func (s *Server) addToTempClients(cid uint64, c *client) bool {
        added := false
        s.grMu.Lock()
        if s.grRunning {
                s.grTmpClients[cid] = c
                added = true
        }
        s.grMu.Unlock()
        return added
}

/////////////////////////////////////////////////////////////////
// These are some helpers for accounting in functional tests.
/////////////////////////////////////////////////////////////////

// NumRoutes will report the number of registered routes.
func (s *Server) NumRoutes() int {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.numRoutes()
}

// numRoutes will report the number of registered routes.
// Server lock held on entry
func (s *Server) numRoutes() int {
        var nr int
        s.forEachRoute(func(c *client) {
                nr++
        })
        return nr
}

// NumRemotes will report number of registered remotes.
func (s *Server) NumRemotes() int {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.numRemotes()
}

// numRemotes will report number of registered remotes.
// Server lock held on entry
func (s *Server) numRemotes() int {
        return len(s.routes)
}

// NumLeafNodes will report number of leaf node connections.
func (s *Server) NumLeafNodes() int {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return len(s.leafs)
}

// NumClients will report the number of registered clients.
func (s *Server) NumClients() int {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return len(s.clients)
}

// GetClient will return the client associated with cid.
func (s *Server) GetClient(cid uint64) *client {
        return s.getClient(cid)
}

// getClient will return the client associated with cid.
func (s *Server) getClient(cid uint64) *client {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.clients[cid]
}

// GetLeafNode returns the leafnode associated with the cid.
func (s *Server) GetLeafNode(cid uint64) *client {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.leafs[cid]
}

// NumSubscriptions will report how many subscriptions are active.
func (s *Server) NumSubscriptions() uint32 {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.numSubscriptions()
}

// numSubscriptions will report how many subscriptions are active.
// Lock should be held.
func (s *Server) numSubscriptions() uint32 {
        var subs int
        s.accounts.Range(func(k, v any) bool {
                acc := v.(*Account)
                subs += acc.TotalSubs()
                return true
        })
        return uint32(subs)
}

// NumSlowConsumers will report the number of slow consumers.
func (s *Server) NumSlowConsumers() int64 {
        return atomic.LoadInt64(&s.slowConsumers)
}

// NumStalledClients will report the total number of times clients have been stalled.
func (s *Server) NumStalledClients() int64 {
        return atomic.LoadInt64(&s.stalls)
}

// NumSlowConsumersClients will report the number of slow consumers clients.
func (s *Server) NumSlowConsumersClients() uint64 {
        return s.scStats.clients.Load()
}

// NumSlowConsumersRoutes will report the number of slow consumers routes.
func (s *Server) NumSlowConsumersRoutes() uint64 {
        return s.scStats.routes.Load()
}

// NumSlowConsumersGateways will report the number of slow consumers leafs.
func (s *Server) NumSlowConsumersGateways() uint64 {
        return s.scStats.gateways.Load()
}

// NumSlowConsumersLeafs will report the number of slow consumers leafs.
func (s *Server) NumSlowConsumersLeafs() uint64 {
        return s.scStats.leafs.Load()
}

// NumStaleConnections will report the number of stale connections.
func (s *Server) NumStaleConnections() int64 {
        return atomic.LoadInt64(&s.staleConnections)
}

// NumStaleConnectionsClients will report the number of stale client connections.
func (s *Server) NumStaleConnectionsClients() uint64 {
        return s.staleStats.clients.Load()
}

// NumStaleConnectionsRoutes will report the number of stale route connections.
func (s *Server) NumStaleConnectionsRoutes() uint64 {
        return s.staleStats.routes.Load()
}

// NumStaleConnectionsGateways will report the number of stale gateway connections.
func (s *Server) NumStaleConnectionsGateways() uint64 {
        return s.staleStats.gateways.Load()
}

// NumStaleConnectionsLeafs will report the number of stale leaf connections.
func (s *Server) NumStaleConnectionsLeafs() uint64 {
        return s.staleStats.leafs.Load()
}

// ConfigTime will report the last time the server configuration was loaded.
func (s *Server) ConfigTime() time.Time {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.configTime
}

// Addr will return the net.Addr object for the current listener.
func (s *Server) Addr() net.Addr {
        s.mu.RLock()
        defer s.mu.RUnlock()
        if s.listener == nil {
                return nil
        }
        return s.listener.Addr()
}

// MonitorAddr will return the net.Addr object for the monitoring listener.
func (s *Server) MonitorAddr() *net.TCPAddr {
        s.mu.RLock()
        defer s.mu.RUnlock()
        if s.http == nil {
                return nil
        }
        return s.http.Addr().(*net.TCPAddr)
}

// ClusterAddr returns the net.Addr object for the route listener.
func (s *Server) ClusterAddr() *net.TCPAddr {
        s.mu.RLock()
        defer s.mu.RUnlock()
        if s.routeListener == nil {
                return nil
        }
        return s.routeListener.Addr().(*net.TCPAddr)
}

// ProfilerAddr returns the net.Addr object for the profiler listener.
func (s *Server) ProfilerAddr() *net.TCPAddr {
        s.mu.RLock()
        defer s.mu.RUnlock()
        if s.profiler == nil {
                return nil
        }
        return s.profiler.Addr().(*net.TCPAddr)
}

func (s *Server) readyForConnections(d time.Duration) error {
        // Snapshot server options.
        opts := s.getOpts()

        type info struct {
                ok  bool
                err error
        }
        chk := make(map[string]info)

        end := time.Now().Add(d)
        for time.Now().Before(end) {
                s.mu.RLock()
                chk["server"] = info{ok: s.listener != nil || opts.DontListen, err: s.listenerErr}
                chk["route"] = info{ok: (opts.Cluster.Port == 0 || s.routeListener != nil), err: s.routeListenerErr}
                chk["gateway"] = info{ok: (opts.Gateway.Name == _EMPTY_ || s.gatewayListener != nil), err: s.gatewayListenerErr}
                chk["leafnode"] = info{ok: (opts.LeafNode.Port == 0 || s.leafNodeListener != nil), err: s.leafNodeListenerErr}
                chk["websocket"] = info{ok: (opts.Websocket.Port == 0 || s.websocket.listener != nil), err: s.websocket.listenerErr}
                chk["mqtt"] = info{ok: (opts.MQTT.Port == 0 || s.mqtt.listener != nil), err: s.mqtt.listenerErr}
                s.mu.RUnlock()

                var numOK int
                for _, inf := range chk {
                        if inf.ok {
                                numOK++
                        }
                }
                if numOK == len(chk) {
                        // In the case of DontListen option (no accept loop), we still want
                        // to make sure that Start() has done all the work, so we wait on
                        // that.
                        if opts.DontListen {
                                select {
                                case <-s.startupComplete:
                                case <-time.After(d):
                                        return fmt.Errorf("failed to be ready for connections after %s: startup did not complete", d)
                                }
                        }
                        return nil
                }
                if d > 25*time.Millisecond {
                        time.Sleep(25 * time.Millisecond)
                }
        }

        failed := make([]string, 0, len(chk))
        for name, inf := range chk {
                if inf.ok && inf.err != nil {
                        failed = append(failed, fmt.Sprintf("%s(ok, but %s)", name, inf.err))
                }
                if !inf.ok && inf.err == nil {
                        failed = append(failed, name)
                }
                if !inf.ok && inf.err != nil {
                        failed = append(failed, fmt.Sprintf("%s(%s)", name, inf.err))
                }
        }

        return fmt.Errorf(
                "failed to be ready for connections after %s: %s",
                d, strings.Join(failed, ", "),
        )
}

// ReadyForConnections returns `true` if the server is ready to accept clients
// and, if routing is enabled, route connections. If after the duration
// `dur` the server is still not ready, returns `false`.
func (s *Server) ReadyForConnections(dur time.Duration) bool {
        return s.readyForConnections(dur) == nil
}

// Quick utility to function to tell if the server supports headers.
func (s *Server) supportsHeaders() bool {
        if s == nil {
                return false
        }
        return !(s.getOpts().NoHeaderSupport)
}

// ID returns the server's ID
func (s *Server) ID() string {
        return s.info.ID
}

// NodeName returns the node name for this server.
func (s *Server) NodeName() string {
        return getHash(s.info.Name)
}

// Name returns the server's name. This will be the same as the ID if it was not set.
func (s *Server) Name() string {
        return s.info.Name
}

func (s *Server) String() string {
        return s.info.Name
}

type pprofLabels map[string]string

func setGoRoutineLabels(tags ...pprofLabels) {
        var labels []string
        for _, m := range tags {
                for k, v := range m {
                        labels = append(labels, k, v)
                }
        }
        if len(labels) > 0 {
                pprof.SetGoroutineLabels(
                        pprof.WithLabels(context.Background(), pprof.Labels(labels...)),
                )
        }
}

func (s *Server) startGoRoutine(f func(), tags ...pprofLabels) bool {
        var started bool
        s.grMu.Lock()
        defer s.grMu.Unlock()
        if s.grRunning {
                s.grWG.Add(1)
                go func() {
                        setGoRoutineLabels(tags...)
                        f()
                }()
                started = true
        }
        return started
}

func (s *Server) numClosedConns() int {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.closed.len()
}

func (s *Server) totalClosedConns() uint64 {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.closed.totalConns()
}

func (s *Server) closedClients() []*closedClient {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.closed.closedClients()
}

// getClientConnectURLs returns suitable URLs for clients to connect to the listen
// port based on the server options' Host and Port. If the Host corresponds to
// "any" interfaces, this call returns the list of resolved IP addresses.
// If ClientAdvertise is set, returns the client advertise host and port.
// The server lock is assumed held on entry.
func (s *Server) getClientConnectURLs() []string {
        // Snapshot server options.
        opts := s.getOpts()
        // Ignore error here since we know that if there is client advertise, the
        // parseHostPort is correct because we did it right before calling this
        // function in Server.New().
        urls, _ := s.getConnectURLs(opts.ClientAdvertise, opts.Host, opts.Port)
        return urls
}

// Generic version that will return an array of URLs based on the given
// advertise, host and port values.
func (s *Server) getConnectURLs(advertise, host string, port int) ([]string, error) {
        urls := make([]string, 0, 1)

        // short circuit if advertise is set
        if advertise != "" {
                h, p, err := parseHostPort(advertise, port)
                if err != nil {
                        return nil, err
                }
                urls = append(urls, net.JoinHostPort(h, strconv.Itoa(p)))
        } else {
                sPort := strconv.Itoa(port)
                _, ips, err := s.getNonLocalIPsIfHostIsIPAny(host, true)
                for _, ip := range ips {
                        urls = append(urls, net.JoinHostPort(ip, sPort))
                }
                if err != nil || len(urls) == 0 {
                        // We are here if s.opts.Host is not "0.0.0.0" nor "::", or if for some
                        // reason we could not add any URL in the loop above.
                        // We had a case where a Windows VM was hosed and would have err == nil
                        // and not add any address in the array in the loop above, and we
                        // ended-up returning 0.0.0.0, which is problematic for Windows clients.
                        // Check for 0.0.0.0 or :: specifically, and ignore if that's the case.
                        if host == "0.0.0.0" || host == "::" {
                                s.Errorf("Address %q can not be resolved properly", host)
                        } else {
                                urls = append(urls, net.JoinHostPort(host, sPort))
                        }
                }
        }
        return urls, nil
}

// Returns an array of non local IPs if the provided host is
// 0.0.0.0 or ::. It returns the first resolved if `all` is
// false.
// The boolean indicate if the provided host was 0.0.0.0 (or ::)
// so that if the returned array is empty caller can decide
// what to do next.
func (s *Server) getNonLocalIPsIfHostIsIPAny(host string, all bool) (bool, []string, error) {
        ip := net.ParseIP(host)
        // If this is not an IP, we are done
        if ip == nil {
                return false, nil, nil
        }
        // If this is not 0.0.0.0 or :: we have nothing to do.
        if !ip.IsUnspecified() {
                return false, nil, nil
        }
        s.Debugf("Get non local IPs for %q", host)
        var ips []string
        ifaces, _ := net.Interfaces()
        for _, i := range ifaces {
                addrs, _ := i.Addrs()
                for _, addr := range addrs {
                        switch v := addr.(type) {
                        case *net.IPNet:
                                ip = v.IP
                        case *net.IPAddr:
                                ip = v.IP
                        }
                        ipStr := ip.String()
                        // Skip non global unicast addresses
                        if !ip.IsGlobalUnicast() || ip.IsUnspecified() {
                                ip = nil
                                continue
                        }
                        s.Debugf("  ip=%s", ipStr)
                        ips = append(ips, ipStr)
                        if !all {
                                break
                        }
                }
        }
        return true, ips, nil
}

// if the ip is not specified, attempt to resolve it
func resolveHostPorts(addr net.Listener) []string {
        hostPorts := make([]string, 0)
        hp := addr.Addr().(*net.TCPAddr)
        port := strconv.Itoa(hp.Port)
        if hp.IP.IsUnspecified() {
                var ip net.IP
                ifaces, _ := net.Interfaces()
                for _, i := range ifaces {
                        addrs, _ := i.Addrs()
                        for _, addr := range addrs {
                                switch v := addr.(type) {
                                case *net.IPNet:
                                        ip = v.IP
                                        hostPorts = append(hostPorts, net.JoinHostPort(ip.String(), port))
                                case *net.IPAddr:
                                        ip = v.IP
                                        hostPorts = append(hostPorts, net.JoinHostPort(ip.String(), port))
                                default:
                                        continue
                                }
                        }
                }
        } else {
                hostPorts = append(hostPorts, net.JoinHostPort(hp.IP.String(), port))
        }
        return hostPorts
}

// format the address of a net.Listener with a protocol
func formatURL(protocol string, addr net.Listener) []string {
        hostports := resolveHostPorts(addr)
        for i, hp := range hostports {
                hostports[i] = fmt.Sprintf("%s://%s", protocol, hp)
        }
        return hostports
}

// Ports describes URLs that the server can be contacted in
type Ports struct {
        Nats       []string `json:"nats,omitempty"`
        Monitoring []string `json:"monitoring,omitempty"`
        Cluster    []string `json:"cluster,omitempty"`
        Profile    []string `json:"profile,omitempty"`
        WebSocket  []string `json:"websocket,omitempty"`
}

// PortsInfo attempts to resolve all the ports. If after maxWait the ports are not
// resolved, it returns nil. Otherwise it returns a Ports struct
// describing ports where the server can be contacted
func (s *Server) PortsInfo(maxWait time.Duration) *Ports {
        if s.readyForListeners(maxWait) {
                opts := s.getOpts()

                s.mu.RLock()
                tls := s.info.TLSRequired
                listener := s.listener
                httpListener := s.http
                clusterListener := s.routeListener
                profileListener := s.profiler
                wsListener := s.websocket.listener
                wss := s.websocket.tls
                s.mu.RUnlock()

                ports := Ports{}

                if listener != nil {
                        natsProto := "nats"
                        if tls {
                                natsProto = "tls"
                        }
                        ports.Nats = formatURL(natsProto, listener)
                }

                if httpListener != nil {
                        monProto := "http"
                        if opts.HTTPSPort != 0 {
                                monProto = "https"
                        }
                        ports.Monitoring = formatURL(monProto, httpListener)
                }

                if clusterListener != nil {
                        clusterProto := "nats"
                        if opts.Cluster.TLSConfig != nil {
                                clusterProto = "tls"
                        }
                        ports.Cluster = formatURL(clusterProto, clusterListener)
                }

                if profileListener != nil {
                        ports.Profile = formatURL("http", profileListener)
                }

                if wsListener != nil {
                        protocol := wsSchemePrefix
                        if wss {
                                protocol = wsSchemePrefixTLS
                        }
                        ports.WebSocket = formatURL(protocol, wsListener)
                }

                return &ports
        }

        return nil
}

// Returns the portsFile. If a non-empty dirHint is provided, the dirHint
// path is used instead of the server option value
func (s *Server) portFile(dirHint string) string {
        dirname := s.getOpts().PortsFileDir
        if dirHint != "" {
                dirname = dirHint
        }
        if dirname == _EMPTY_ {
                return _EMPTY_
        }
        return filepath.Join(dirname, fmt.Sprintf("%s_%d.ports", filepath.Base(os.Args[0]), os.Getpid()))
}

// Delete the ports file. If a non-empty dirHint is provided, the dirHint
// path is used instead of the server option value
func (s *Server) deletePortsFile(hintDir string) {
        portsFile := s.portFile(hintDir)
        if portsFile != "" {
                if err := os.Remove(portsFile); err != nil {
                        s.Errorf("Error cleaning up ports file %s: %v", portsFile, err)
                }
        }
}

// Writes a file with a serialized Ports to the specified ports_file_dir.
// The name of the file is `exename_pid.ports`, typically nats-server_pid.ports.
// if ports file is not set, this function has no effect
func (s *Server) logPorts() {
        opts := s.getOpts()
        portsFile := s.portFile(opts.PortsFileDir)
        if portsFile != _EMPTY_ {
                go func() {
                        info := s.PortsInfo(5 * time.Second)
                        if info == nil {
                                s.Errorf("Unable to resolve the ports in the specified time")
                                return
                        }
                        data, err := json.Marshal(info)
                        if err != nil {
                                s.Errorf("Error marshaling ports file: %v", err)
                                return
                        }
                        if err := os.WriteFile(portsFile, data, 0666); err != nil {
                                s.Errorf("Error writing ports file (%s): %v", portsFile, err)
                                return
                        }

                }()
        }
}

// waits until a calculated list of listeners is resolved or a timeout
func (s *Server) readyForListeners(dur time.Duration) bool {
        end := time.Now().Add(dur)
        for time.Now().Before(end) {
                s.mu.RLock()
                listeners := s.serviceListeners()
                s.mu.RUnlock()
                if len(listeners) == 0 {
                        return false
                }

                ok := true
                for _, l := range listeners {
                        if l == nil {
                                ok = false
                                break
                        }
                }
                if ok {
                        return true
                }
                select {
                case <-s.quitCh:
                        return false
                case <-time.After(25 * time.Millisecond):
                        // continue - unable to select from quit - we are still running
                }
        }
        return false
}

// returns a list of listeners that are intended for the process
// if the entry is nil, the interface is yet to be resolved
func (s *Server) serviceListeners() []net.Listener {
        listeners := make([]net.Listener, 0)
        opts := s.getOpts()
        listeners = append(listeners, s.listener)
        if opts.Cluster.Port != 0 {
                listeners = append(listeners, s.routeListener)
        }
        if opts.HTTPPort != 0 || opts.HTTPSPort != 0 {
                listeners = append(listeners, s.http)
        }
        if opts.ProfPort != 0 {
                listeners = append(listeners, s.profiler)
        }
        if opts.Websocket.Port != 0 {
                listeners = append(listeners, s.websocket.listener)
        }
        return listeners
}

// Returns true if in lame duck mode.
func (s *Server) isLameDuckMode() bool {
        s.mu.RLock()
        defer s.mu.RUnlock()
        return s.ldm
}

// LameDuckShutdown will perform a lame duck shutdown of NATS, whereby
// the client listener is closed, existing client connections are
// kicked, Raft leaderships are transferred, JetStream is shutdown
// and then finally shutdown the the NATS Server itself.
// This function blocks and will not return until the NATS Server
// has completed the entire shutdown operation.
func (s *Server) LameDuckShutdown() {
        s.lameDuckMode()
}

// This function will close the client listener then close the clients
// at some interval to avoid a reconnect storm.
// We will also transfer any raft leaders and shutdown JetStream.
func (s *Server) lameDuckMode() {
        s.mu.Lock()
        // Check if there is actually anything to do
        if s.isShuttingDown() || s.ldm || s.listener == nil {
                s.mu.Unlock()
                return
        }
        s.Noticef("Entering lame duck mode, stop accepting new clients")
        s.ldm = true
        s.sendLDMShutdownEventLocked()
        expected := 1
        s.listener.Close()
        s.listener = nil
        expected += s.closeWebsocketServer()
        s.ldmCh = make(chan bool, expected)
        opts := s.getOpts()
        gp := opts.LameDuckGracePeriod
        // For tests, we want the grace period to be in some cases bigger
        // than the ldm duration, so to by-pass the validateOptions() check,
        // we use negative number and flip it here.
        if gp < 0 {
                gp *= -1
        }
        s.mu.Unlock()

        // If we are running any raftNodes transfer leaders.
        if hadTransfers := s.transferRaftLeaders(); hadTransfers {
                // They will transfer leadership quickly, but wait here for a second.
                select {
                case <-time.After(time.Second):
                case <-s.quitCh:
                        return
                }
        }

        // Now check and shutdown jetstream.
        s.shutdownJetStream()

        // Now shutdown the nodes
        s.shutdownRaftNodes()

        // Wait for accept loops to be done to make sure that no new
        // client can connect
        for i := 0; i < expected; i++ {
                <-s.ldmCh
        }

        s.mu.Lock()
        // Need to recheck few things
        if s.isShuttingDown() || len(s.clients) == 0 {
                s.mu.Unlock()
                // If there is no client, we need to call Shutdown() to complete
                // the LDMode. If server has been shutdown while lock was released,
                // calling Shutdown() should be no-op.
                s.Shutdown()
                return
        }
        dur := int64(opts.LameDuckDuration)
        dur -= int64(gp)
        if dur <= 0 {
                dur = int64(time.Second)
        }
        numClients := int64(len(s.clients))
        batch := 1
        // Sleep interval between each client connection close.
        var si int64
        if numClients != 0 {
                si = dur / numClients
        }
        if si < 1 {
                // Should not happen (except in test with very small LD duration), but
                // if there are too many clients, batch the number of close and
                // use a tiny sleep interval that will result in yield likely.
                si = 1
                batch = int(numClients / dur)
        } else if si > int64(time.Second) {
                // Conversely, there is no need to sleep too long between clients
                // and spread say 10 clients for the 2min duration. Sleeping no
                // more than 1sec.
                si = int64(time.Second)
        }

        // Now capture all clients
        clients := make([]*client, 0, len(s.clients))
        for _, client := range s.clients {
                clients = append(clients, client)
        }
        // Now that we know that no new client can be accepted,
        // send INFO to routes and clients to notify this state.
        s.sendLDMToRoutes()
        s.sendLDMToClients()
        s.mu.Unlock()

        t := time.NewTimer(gp)
        // Delay start of closing of client connections in case
        // we have several servers that we want to signal to enter LD mode
        // and not have their client reconnect to each other.
        select {
        case <-t.C:
                s.Noticef("Closing existing clients")
        case <-s.quitCh:
                t.Stop()
                return
        }
        for i, client := range clients {
                client.closeConnection(ServerShutdown)
                if i == len(clients)-1 {
                        break
                }
                if batch == 1 || i%batch == 0 {
                        // We pick a random interval which will be at least si/2
                        v := rand.Int63n(si)
                        if v < si/2 {
                                v = si / 2
                        }
                        t.Reset(time.Duration(v))
                        // Sleep for given interval or bail out if kicked by Shutdown().
                        select {
                        case <-t.C:
                        case <-s.quitCh:
                                t.Stop()
                                return
                        }
                }
        }
        s.Shutdown()
        s.WaitForShutdown()
}

// Send an INFO update to routes with the indication that this server is in LDM mode.
// Server lock is held on entry.
func (s *Server) sendLDMToRoutes() {
        s.routeInfo.LameDuckMode = true
        infoJSON := generateInfoJSON(&s.routeInfo)
        s.forEachRemote(func(r *client) {
                r.mu.Lock()
                r.enqueueProto(infoJSON)
                r.mu.Unlock()
        })
        // Clear now so that we notify only once, should we have to send other INFOs.
        s.routeInfo.LameDuckMode = false
}

// Send an INFO update to clients with the indication that this server is in
// LDM mode and with only URLs of other nodes.
// Server lock is held on entry.
func (s *Server) sendLDMToClients() {
        s.info.LameDuckMode = true
        // Clear this so that if there are further updates, we don't send our URLs.
        s.clientConnectURLs = s.clientConnectURLs[:0]
        if s.websocket.connectURLs != nil {
                s.websocket.connectURLs = s.websocket.connectURLs[:0]
        }
        // Reset content first.
        s.info.ClientConnectURLs = s.info.ClientConnectURLs[:0]
        s.info.WSConnectURLs = s.info.WSConnectURLs[:0]
        // Only add the other nodes if we are allowed to.
        if !s.getOpts().Cluster.NoAdvertise {
                for url := range s.clientConnectURLsMap {
                        s.info.ClientConnectURLs = append(s.info.ClientConnectURLs, url)
                }
                for url := range s.websocket.connectURLsMap {
                        s.info.WSConnectURLs = append(s.info.WSConnectURLs, url)
                }
        }
        // Send to all registered clients that support async INFO protocols.
        s.sendAsyncInfoToClients(true, true)
        // We now clear the info.LameDuckMode flag so that if there are
        // cluster updates and we send the INFO, we don't have the boolean
        // set which would cause multiple LDM notifications to clients.
        s.info.LameDuckMode = false
}

// If given error is a net.Error and is temporary, sleeps for the given
// delay and double it, but cap it to ACCEPT_MAX_SLEEP. The sleep is
// interrupted if the server is shutdown.
// An error message is displayed depending on the type of error.
// Returns the new (or unchanged) delay, or a negative value if the
// server has been or is being shutdown.
func (s *Server) acceptError(acceptName string, err error, tmpDelay time.Duration) time.Duration {
        if !s.isRunning() {
                return -1
        }
        //lint:ignore SA1019 We want to retry on a bunch of errors here.
        if ne, ok := err.(net.Error); ok && ne.Temporary() { // nolint:staticcheck
                s.Errorf("Temporary %s Accept Error(%v), sleeping %dms", acceptName, ne, tmpDelay/time.Millisecond)
                select {
                case <-time.After(tmpDelay):
                case <-s.quitCh:
                        return -1
                }
                tmpDelay *= 2
                if tmpDelay > ACCEPT_MAX_SLEEP {
                        tmpDelay = ACCEPT_MAX_SLEEP
                }
        } else {
                s.Errorf("%s Accept error: %v", acceptName, err)
        }
        return tmpDelay
}

var errNoIPAvail = errors.New("no IP available")

func (s *Server) getRandomIP(resolver netResolver, url string, excludedAddresses map[string]struct{}) (string, error) {
        host, port, err := net.SplitHostPort(url)
        if err != nil {
                return "", err
        }
        // If already an IP, skip.
        if net.ParseIP(host) != nil {
                return url, nil
        }
        ips, err := resolver.LookupHost(context.Background(), host)
        if err != nil {
                return "", fmt.Errorf("lookup for host %q: %v", host, err)
        }
        if len(excludedAddresses) > 0 {
                for i := 0; i < len(ips); i++ {
                        ip := ips[i]
                        addr := net.JoinHostPort(ip, port)
                        if _, excluded := excludedAddresses[addr]; excluded {
                                if len(ips) == 1 {
                                        ips = nil
                                        break
                                }
                                ips[i] = ips[len(ips)-1]
                                ips = ips[:len(ips)-1]
                                i--
                        }
                }
                if len(ips) == 0 {
                        return "", errNoIPAvail
                }
        }
        var address string
        if len(ips) == 0 {
                s.Warnf("Unable to get IP for %s, will try with %s: %v", host, url, err)
                address = url
        } else {
                var ip string
                if len(ips) == 1 {
                        ip = ips[0]
                } else {
                        ip = ips[rand.Int31n(int32(len(ips)))]
                }
                // add the port
                address = net.JoinHostPort(ip, port)
        }
        return address, nil
}

// Returns true for the first attempt and depending on the nature
// of the attempt (first connect or a reconnect), when the number
// of attempts is equal to the configured report attempts.
func (s *Server) shouldReportConnectErr(firstConnect bool, attempts int) bool {
        opts := s.getOpts()
        if firstConnect {
                if attempts == 1 || attempts%opts.ConnectErrorReports == 0 {
                        return true
                }
                return false
        }
        if attempts == 1 || attempts%opts.ReconnectErrorReports == 0 {
                return true
        }
        return false
}

func (s *Server) updateRemoteSubscription(acc *Account, sub *subscription, delta int32) {
        s.updateRouteSubscriptionMap(acc, sub, delta)
        if s.gateway.enabled {
                s.gatewayUpdateSubInterest(acc.Name, sub, delta)
        }

        acc.updateLeafNodes(sub, delta)
}

func (s *Server) startRateLimitLogExpiration() {
        interval := time.Second
        s.startGoRoutine(func() {
                defer s.grWG.Done()

                ticker := time.NewTicker(time.Second)
                defer ticker.Stop()
                for {
                        select {
                        case <-s.quitCh:
                                return
                        case interval = <-s.rateLimitLoggingCh:
                                ticker.Reset(interval)
                        case <-ticker.C:
                                s.rateLimitLogging.Range(func(k, v any) bool {
                                        start := v.(time.Time)
                                        if time.Since(start) >= interval {
                                                s.rateLimitLogging.Delete(k)
                                        }
                                        return true
                                })
                        }
                }
        })
}

func (s *Server) changeRateLimitLogInterval(d time.Duration) {
        if d <= 0 {
                return
        }
        select {
        case s.rateLimitLoggingCh <- d:
        default:
        }
}

// DisconnectClientByID disconnects a client by connection ID
func (s *Server) DisconnectClientByID(id uint64) error {
        if s == nil {
                return ErrServerNotRunning
        }
        if client := s.getClient(id); client != nil {
                client.closeConnection(Kicked)
                return nil
        } else if client = s.GetLeafNode(id); client != nil {
                client.closeConnection(Kicked)
                return nil
        }
        return errors.New("no such client or leafnode id")
}

// LDMClientByID sends a Lame Duck Mode info message to a client by connection ID
func (s *Server) LDMClientByID(id uint64) error {
        if s == nil {
                return ErrServerNotRunning
        }
        s.mu.RLock()
        c := s.clients[id]
        if c == nil {
                s.mu.RUnlock()
                return errors.New("no such client id")
        }
        info := s.copyInfo()
        info.LameDuckMode = true
        s.mu.RUnlock()
        c.mu.Lock()
        defer c.mu.Unlock()
        if c.opts.Protocol >= ClientProtoInfo && c.flags.isSet(firstPongSent) {
                // sendInfo takes care of checking if the connection is still
                // valid or not, so don't duplicate tests here.
                c.Debugf("Sending Lame Duck Mode info to client")
                c.enqueueProto(c.generateClientInfoJSON(info))
                return nil
        } else {
                return errors.New("client does not support Lame Duck Mode or is not ready to receive the notification")
        }
}

nats-io / nats-server / 19188141845

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous