24949216239

Committed 24 Apr 2026 08:34AM UTC coverage: 80.645% (-2.4%) from 83.05%

Build # 24949216239

Build Type

push

github

Committed by

web-flow

Commit Message

(2.14) [ADDED] `RemoteLeafOpts.IgnoreDiscoveredServers` option (#8067)

For a given leafnode remote, if this is set to true, this remote will
ignore any server leafnode URLs returned by the hub, allowing the user
to fully manage the servers this remote can connect to.

Resolves #8002

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

Coverage Stats

74685 of 92610 relevant lines covered (80.64%)

632737.46 hits per line

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

88.73

/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"
        "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     map[*leafNodeCfg]struct{}
        rmLeafRemoteCfgs   map[string]*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]
        jsAPIRoutedInfoReqs *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
}

// 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)
        s.printFeatureFlags(opts)

        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 if PROXY protocol is enabled
        if opts.ProxyProtocol {
                s.Noticef("PROXY protocol enabled for client connections")
        }

        // 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.
        // If we have ProxyProtocol enabled then we won't include the client
        // IP in the initial INFO, as that would leak the proxy IP itself.
        // In that case we'll send another INFO after the client introduces itself.
        tlsRequired := info.TLSRequired
        infoBytes := c.generateClientInfoJSON(info, !opts.ProxyProtocol)

        // 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 || (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, !opts.ProxyProtocol)
                        }
                        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, 6) // Minimum 6 bytes for proxy proto in next step.
                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 proxy protocol if enabled.
        if !isClosed && !tlsRequired && opts.ProxyProtocol {
                if len(pre) == 0 {
                        // There has been no pre-read yet, do so so we can work out
                        // if the client is trying to negotiate PROXY.
                        pre = make([]byte, 6)
                        c.nc.SetReadDeadline(time.Now().Add(proxyProtoReadTimeout))
                        n, _ := io.ReadFull(c.nc, pre)
                        c.nc.SetReadDeadline(time.Time{})
                        pre = pre[:n]
                }
                conn = &tlsMixConn{conn, bytes.NewBuffer(pre)}
                addr, proxyPre, err := readProxyProtoHeader(conn)
                if err != nil && err != errProxyProtoUnrecognized {
                        // err != errProxyProtoUnrecognized implies that we detected a proxy
                        // protocol header but we failed to parse it, so don't continue.
                        c.mu.Unlock()
                        s.Warnf("Error reading PROXY protocol header from %s: %v", conn.RemoteAddr(), err)
                        c.closeConnection(ProtocolViolation)
                        return nil
                }
                // If addr is nil, it was a LOCAL/UNKNOWN command (health check)
                // Use the connection as-is
                if addr != nil {
                        c.nc = &proxyConn{
                                Conn:       conn,
                                remoteAddr: addr,
                        }
                        // These were set already by initClient, override them.
                        c.host = addr.srcIP.String()
                        c.port = addr.srcPort
                }
                // At this point, err is either:
                //  - nil => we parsed the proxy protocol header successfully
                //  - errProxyProtoUnrecognized => we didn't detect proxy protocol at all
                // We only clear the pre-read if we successfully read the protocol header
                // so that the next step doesn't re-read it. Otherwise we have to assume
                // that it's a non-proxied connection and we want the pre-read to remain
                // for the next step.
                if err == nil {
                        pre = proxyPre
                }
                // Because we have ProxyProtocol enabled, our earlier INFO message didn't
                // include the client_ip. If we need to send it again then we will include
                // it, but sending it here immediately can confuse clients who have just
                // PING'd.
                infoBytes = c.generateClientInfoJSON(info, true)
        }

        // 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, true))
                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 / 24949216239

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