jetstream_helpers_test.go

// Copyright 2020-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.

// Do not exlude this file with the !skip_js_tests since those helpers
// are also used by MQTT.

//lint:file-ignore U1000 Avoid detecting as unused code

package server

import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"io/fs"
"math/rand"
"net"
"net/url"
"os"
"path"
"strings"
"sync"
"testing"
"time"

"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"golang.org/x/time/rate"

"github.com/nats-io/nats-server/v2/internal/antithesis"
)

// Support functions

funcinit() {
// Speed up raft for tests.
hbInterval=50*time.Millisecond
minElectionTimeout=1500*time.Millisecond
maxElectionTimeout=3500*time.Millisecond
lostQuorumInterval=2*time.Second
lostQuorumCheck=4*hbInterval

// For statz and jetstream placement speedups as well.
statszRateLimit=0
}

// Used to setup clusters of clusters for tests.
typeclusterstruct {
servers []*Server
opts []*Options
namestring
t testing.TB
nproxies []*netProxy
}

// Used to setup superclusters for tests.
typesuperclusterstruct {
t*testing.T
clusters []*cluster
nproxies []*netProxy
}

func (sc*supercluster) shutdown() {
ifsc==nil {
return
 }
for_, np:=rangesc.nproxies {
np.stop()
 }
for_, c:=rangesc.clusters {
shutdownCluster(c)
 }
}

func (sc*supercluster) randomServer() *Server {
returnsc.randomCluster().randomServer()
}

func (sc*supercluster) serverByName(snamestring) *Server {
for_, c:=rangesc.clusters {
ifs:=c.serverByName(sname); s!=nil {
returns
 }
 }
returnnil
}

func (sc*supercluster) waitOnStreamLeader(account, streamstring) {
sc.t.Helper()
expires:=time.Now().Add(30*time.Second)
fortime.Now().Before(expires) {
for_, c:=rangesc.clusters {
ifleader:=c.streamLeader(account, stream); leader!=nil {
time.Sleep(200*time.Millisecond)
return
 }
 }
time.Sleep(100*time.Millisecond)
 }
antithesis.AssertUnreachable(sc.t, "Timeout in supercluster.waitOnStreamLeader", map[string]any{
"account": account,
"stream": stream,
 })
sc.t.Fatalf("Expected a stream leader for %q %q, got none", account, stream)
}

varjsClusterAccountsTempl=`
 listen: 127.0.0.1:-1

 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 leaf {
 listen: 127.0.0.1:-1
 }

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 websocket {
 listen: 127.0.0.1:-1
 compression: true
 handshake_timeout: "5s"
 no_tls: true
 }

 no_auth_user: one

 accounts {
 ONE { users = [ { user: "one", pass: "p" } ]; jetstream: enabled }
 TWO { users = [ { user: "two", pass: "p" } ]; jetstream: enabled }
 NOJS { users = [ { user: "nojs", pass: "p" } ] }
 $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 }
`

varjsClusterTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 leaf {
 listen: 127.0.0.1:-1
 }

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 # For access to system account.
 accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
`

varjsClusterEncryptedTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s', key: "s3cr3t!"}

 leaf {
 listen: 127.0.0.1:-1
 }

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 # For access to system account.
 accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
`

varjsClusterMaxBytesTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 leaf {
 listen: 127.0.0.1:-1
 }

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 no_auth_user: u

 accounts {
 $U {
 users = [ { user: "u", pass: "p" } ]
 jetstream: {
 max_mem: 128MB
 max_file: 18GB
 max_bytes: true // Forces streams to indicate max_bytes.
 }
 }
 $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 }
`

varjsClusterMaxBytesAccountLimitTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 256MB, max_file_store: 4GB, store_dir: '%s'}

 leaf {
 listen: 127.0.0.1:-1
 }

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 no_auth_user: u

 accounts {
 $U {
 users = [ { user: "u", pass: "p" } ]
 jetstream: {
 max_mem: 128MB
 max_file: 3GB
 max_bytes: true // Forces streams to indicate max_bytes.
 }
 }
 $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 }
`

varjsSuperClusterTempl=`
 %s
 gateway {
 name: %s
 listen: 127.0.0.1:%d
 gateways = [%s
 ]
 }

 system_account: "$SYS"
`

varjsClusterLimitsTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2MB, max_file_store: 8MB, store_dir: '%s'}

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 no_auth_user: u

 accounts {
 ONE {
 users = [ { user: "u", pass: "s3cr3t!" } ]
 jetstream: enabled
 }
 $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 }
`

varjsMixedModeGlobalAccountTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2MB, max_file_store: 8MB, store_dir: '%s'}

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 accounts {$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
`

varjsGWTempl=`%s{name: %s, urls: [%s]}`

varjsClusterAccountLimitsTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 no_auth_user: js

 accounts {
 $JS { users = [ { user: "js", pass: "p" } ]; jetstream: {max_store: 1MB, max_mem: 0} }
 $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 }
`

funccreateJetStreamTaggedSuperCluster(t*testing.T) *supercluster {
returncreateJetStreamTaggedSuperClusterWithGWProxy(t, nil)
}

funccreateJetStreamTaggedSuperClusterWithGWProxy(t*testing.T, gwmgwProxyMap) *supercluster {
sc:=createJetStreamSuperClusterWithTemplateAndModHook(t, jsClusterTempl, 3, 3, nil, gwm)
sc.waitOnPeerCount(9)

reset:=func(s*Server) {
s.mu.Lock()
rch:=s.sys.resetCh
s.mu.Unlock()
ifrch!=nil {
rch<-struct{}{}
 }
s.sendStatszUpdate()
 }

// Make first cluster AWS, US country code.
fori, s:=rangesc.clusterForName("C1").servers {
s.optsMu.Lock()
s.opts.Tags.Add("cloud:aws")
s.opts.Tags.Add("country:us")
s.opts.Tags.Add(fmt.Sprintf("node:%d", i+1))
s.optsMu.Unlock()
reset(s)
 }
// Make second cluster GCP, UK country code.
fori, s:=rangesc.clusterForName("C2").servers {
s.optsMu.Lock()
s.opts.Tags.Add("cloud:gcp")
s.opts.Tags.Add("country:uk")
s.opts.Tags.Add(fmt.Sprintf("node:%d", i+1))
s.optsMu.Unlock()
reset(s)
 }
// Make third cluster AZ, JP country code.
fori, s:=rangesc.clusterForName("C3").servers {
s.optsMu.Lock()
s.opts.Tags.Add("cloud:az")
s.opts.Tags.Add("country:jp")
s.opts.Tags.Add(fmt.Sprintf("node:%d", i+1))
s.optsMu.Unlock()
reset(s)
 }

sc.waitOnLeader()
ml:=sc.leader()
require_True(t, ml!=nil)
js:=ml.getJetStream()
require_True(t, js!=nil)
js.mu.RLock()
deferjs.mu.RUnlock()
cc:=js.cluster
require_True(t, cc!=nil)

// Walk and make sure all tags are registered.
expires:=time.Now().Add(10*time.Second)
fortime.Now().Before(expires) {
allOK:=true
for_, p:=rangecc.meta.Peers() {
si, ok:=ml.nodeToInfo.Load(p.ID)
require_True(t, ok)
ni:=si.(nodeInfo)
iflen(ni.tags) ==0 {
allOK=false
reset(sc.serverByName(ni.name))
 }
 }
ifallOK {
break
 }
 }

returnsc
}

funccreateJetStreamSuperCluster(t*testing.T, numServersPer, numClustersint) *supercluster {
returncreateJetStreamSuperClusterWithTemplate(t, jsClusterTempl, numServersPer, numClusters)
}

funccreateJetStreamSuperClusterWithTemplate(t*testing.T, tmplstring, numServersPer, numClustersint) *supercluster {
returncreateJetStreamSuperClusterWithTemplateAndModHook(t, tmpl, numServersPer, numClusters, nil, nil)
}

// For doing proxyies in GWs.
typegwProxystruct {
rtt time.Duration
upint
downint
}

// For use in normal clusters.
typeclusterProxy=gwProxy

// Maps cluster names to proxy settings.
typegwProxyMapmap[string]*gwProxy

funccreateJetStreamSuperClusterWithTemplateAndModHook(t*testing.T, tmplstring, numServersPer, numClustersint, modifymodifyCb, gwmgwProxyMap) *supercluster {
t.Helper()
ifnumServersPer<1 {
t.Fatalf("Number of servers must be >= 1")
 }
ifnumClusters<=1 {
t.Fatalf("Number of clusters must be > 1")
 }

startClusterPorts:= []int{20_022, 22_022, 24_022}
startGatewayPorts:= []int{20_122, 22_122, 24_122}
startClusterPort:=startClusterPorts[rand.Intn(len(startClusterPorts))]
startGWPort:=startGatewayPorts[rand.Intn(len(startGatewayPorts))]

// Make the GWs form faster for the tests.
SetGatewaysSolicitDelay(10*time.Millisecond)
deferResetGatewaysSolicitDelay()

cp, gp:=startClusterPort, startGWPort
varclusters []*cluster
varnproxies []*netProxy
vargws []string

// Build GWs first, will be same for all servers.
fori, port:=1, gp; i<=numClusters; i++ {
cn:=fmt.Sprintf("C%d", i)
vargwp*gwProxy
iflen(gwm) >0 {
gwp=gwm[cn]
 }
varurls []string
forn:=0; n<numServersPer; n++ {
routeURL:=fmt.Sprintf("nats-route://127.0.0.1:%d", port)
ifgwp!=nil {
np:=createNetProxy(gwp.rtt, gwp.up, gwp.down, routeURL, false)
nproxies=append(nproxies, np)
routeURL=np.routeURL()
 }
urls=append(urls, routeURL)
port++
 }
gws=append(gws, fmt.Sprintf(jsGWTempl, "\n\t\t\t", cn, strings.Join(urls, ",")))
 }
gwconf:=strings.Join(gws, _EMPTY_)

fori:=1; i<=numClusters; i++ {
cn:=fmt.Sprintf("C%d", i)
// Go ahead and build configurations.
c:=&cluster{servers: make([]*Server, 0, numServersPer), opts: make([]*Options, 0, numServersPer), name: cn}

// Build out the routes that will be shared with all configs.
varroutes []string
forport:=cp; port<cp+numServersPer; port++ {
routes=append(routes, fmt.Sprintf("nats-route://127.0.0.1:%d", port))
 }
routeConfig:=strings.Join(routes, ",")

forsi:=0; si<numServersPer; si++ {
storeDir:=t.TempDir()
sn:=fmt.Sprintf("%s-S%d", cn, si+1)
bconf:=fmt.Sprintf(tmpl, sn, storeDir, cn, cp+si, routeConfig)
conf:=fmt.Sprintf(jsSuperClusterTempl, bconf, cn, gp, gwconf)
gp++
ifmodify!=nil {
conf=modify(sn, cn, storeDir, conf)
 }
s, o:=RunServerWithConfig(createConfFile(t, []byte(conf)))
c.servers=append(c.servers, s)
c.opts=append(c.opts, o)
 }
checkClusterFormed(t, c.servers...)
clusters=append(clusters, c)
cp+=numServersPer
c.t=t
 }

// Start any proxies.
for_, np:=rangenproxies {
np.start()
 }

// Wait for the supercluster to be formed.
egws:=numClusters-1
for_, c:=rangeclusters {
for_, s:=rangec.servers {
waitForOutboundGateways(t, s, egws, 10*time.Second)
 }
 }

sc:=&supercluster{t, clusters, nproxies}
sc.waitOnLeader()
sc.waitOnAllCurrent()

// Wait for all the peer nodes to be registered.
checkFor(t, 5*time.Second, 100*time.Millisecond, func() error {
varpeers []string
ifml:=sc.leader(); ml!=nil {
peers=ml.ActivePeers()
iflen(peers) ==numClusters*numServersPer {
returnnil
 }
 }
returnfmt.Errorf("Not correct number of peers, expected %d, got %d", numClusters*numServersPer, len(peers))
 })

ifsc.leader() ==nil {
sc.t.Fatalf("Expected a cluster leader, got none")
 }

returnsc
}

func (sc*supercluster) createLeafNodes(clusterNamestring, numServersint) *cluster {
sc.t.Helper()

// Create our leafnode cluster template first.
returnsc.createLeafNodesWithDomain(clusterName, numServers, "")
}

func (sc*supercluster) createLeafNodesWithDomain(clusterNamestring, numServersint, domainstring) *cluster {
sc.t.Helper()

// Create our leafnode cluster template first.
returnsc.randomCluster().createLeafNodes(clusterName, numServers, domain)
}

func (sc*supercluster) createSingleLeafNode(extendbool) *Server {
sc.t.Helper()

returnsc.randomCluster().createLeafNode(extend)
}

func (sc*supercluster) leader() *Server {
for_, c:=rangesc.clusters {
ifleader:=c.leader(); leader!=nil {
returnleader
 }
 }
returnnil
}

func (sc*supercluster) waitOnLeader() {
sc.t.Helper()
expires:=time.Now().Add(30*time.Second)
fortime.Now().Before(expires) {
for_, c:=rangesc.clusters {
ifleader:=c.leader(); leader!=nil {
time.Sleep(250*time.Millisecond)
return
 }
 }
time.Sleep(25*time.Millisecond)
 }
antithesis.AssertUnreachable(sc.t, "Timeout in supercluster.waitOnStreamLeader", nil)
sc.t.Fatalf("Expected a cluster leader, got none")
}

func (sc*supercluster) waitOnAccount(accountstring) {
sc.t.Helper()
expires:=time.Now().Add(40*time.Second)
fortime.Now().Before(expires) {
found:=true
for_, c:=rangesc.clusters {
for_, s:=rangec.servers {
acc, err:=s.fetchAccount(account)
found=found&&err==nil&&acc!=nil
 }
 }
iffound {
return
 }
time.Sleep(100*time.Millisecond)
continue
 }

sc.t.Fatalf("Expected account %q to exist but didn't", account)
}

func (sc*supercluster) waitOnAllCurrent() {
sc.t.Helper()
for_, c:=rangesc.clusters {
c.waitOnAllCurrent()
 }
}

func (sc*supercluster) clusterForName(namestring) *cluster {
for_, c:=rangesc.clusters {
ifc.name==name {
returnc
 }
 }
returnnil
}

func (sc*supercluster) randomCluster() *cluster {
clusters:=append(sc.clusters[:0:0], sc.clusters...)
rand.Shuffle(len(clusters), func(i, jint) { clusters[i], clusters[j] =clusters[j], clusters[i] })
returnclusters[0]
}

func (sc*supercluster) waitOnPeerCount(nint) {
sc.t.Helper()
sc.waitOnLeader()
leader:=sc.leader()
expires:=time.Now().Add(30*time.Second)
// Make sure we have all peers, and take into account the meta leader could still change.
fortime.Now().Before(expires) {
ifleader=sc.leader(); leader==nil {
time.Sleep(50*time.Millisecond)
continue
 }
iflen(leader.JetStreamClusterPeers()) ==n {
return
 }
time.Sleep(100*time.Millisecond)
 }
antithesis.AssertUnreachable(sc.t, "Timeout in supercluster.waitOnPeerCount", nil)
sc.t.Fatalf("Expected a super cluster peer count of %d, got %d", n, len(leader.JetStreamClusterPeers()))
}

varjsClusterMirrorSourceImportsTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 no_auth_user: dlc

 accounts {
 JS {
 jetstream: enabled
 users = [ { user: "rip", pass: "pass" } ]
 exports [
 { service: "$JS.API.CONSUMER.>" } # To create internal consumers to mirror/source.
 { stream: "RI.DELIVER.SYNC.>" } # For the mirror/source consumers sending to IA via delivery subject.
 { service: "$JS.FC.>" }
 ]
 }
 IA {
 jetstream: enabled
 users = [ { user: "dlc", pass: "pass" } ]
 imports [
 { service: { account: JS, subject: "$JS.API.CONSUMER.>"}, to: "RI.JS.API.CONSUMER.>" }
 { stream: { account: JS, subject: "RI.DELIVER.SYNC.>"} }
 { service: {account: JS, subject: "$JS.FC.>" }}
 ]
 }
 $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 }
`

varjsClusterImportsTempl=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 no_auth_user: dlc

 accounts {
 JS {
 jetstream: enabled
 users = [ { user: "rip", pass: "pass" } ]
 exports [
 { service: "$JS.API.>", response: stream }
 { service: "TEST" } # For publishing to the stream.
 { service: "$JS.ACK.TEST.*.>" }
 ]
 }
 IA {
 users = [ { user: "dlc", pass: "pass" } ]
 imports [
 { service: { subject: "$JS.API.>", account: JS }}
 { service: { subject: "TEST", account: JS }}
 { service: { subject: "$JS.ACK.TEST.*.>", account: JS }}
 ]
 }
 $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 }
`

funccreateMixedModeCluster(t testing.TB, tmplstring, clusterName, snPrestring, numJsServers, numNonServersint, doJSConfigbool) *cluster {
t.Helper()

ifclusterName==_EMPTY_||numJsServers<0||numNonServers<1 {
t.Fatalf("Bad params")
 }

numServers:=numJsServers+numNonServers
conststartClusterPort=23232

// Build out the routes that will be shared with all configs.
varroutes []string
forcp:=startClusterPort; cp<startClusterPort+numServers; cp++ {
routes=append(routes, fmt.Sprintf("nats-route://127.0.0.1:%d", cp))
 }
routeConfig:=strings.Join(routes, ",")

// Go ahead and build configurations and start servers.
c:=&cluster{servers: make([]*Server, 0, numServers), opts: make([]*Options, 0, numServers), name: clusterName}

forcp:=startClusterPort; cp<startClusterPort+numServers; cp++ {
storeDir:=t.TempDir()

sn:=fmt.Sprintf("%sS-%d", snPre, cp-startClusterPort+1)
conf:=fmt.Sprintf(tmpl, sn, storeDir, clusterName, cp, routeConfig)

// Disable JS here.
ifcp-startClusterPort>=numJsServers {
// We can disable by commmenting it out, meaning no JS config, or can set the config up and just set disabled.
// e.g. jetstream: {domain: "SPOKE", enabled: false}
ifdoJSConfig {
conf=strings.Replace(conf, "jetstream: {", "jetstream: { enabled: false, ", 1)
 } else {
conf=strings.Replace(conf, "jetstream: ", "# jetstream: ", 1)
 }
 }

s, o:=RunServerWithConfig(createConfFile(t, []byte(conf)))
c.servers=append(c.servers, s)
c.opts=append(c.opts, o)
 }
c.t=t

// Wait til we are formed and have a leader.
c.checkClusterFormed()
ifnumJsServers>0 {
c.waitOnPeerCount(numJsServers)
 }

returnc
}

// This will create a cluster that is explicitly configured for the routes, etc.
// and also has a defined clustername. All configs for routes and cluster name will be the same.
funccreateJetStreamClusterExplicit(t testing.TB, clusterNamestring, numServersint) *cluster {
returncreateJetStreamClusterWithTemplate(t, jsClusterTempl, clusterName, numServers)
}

funccreateJetStreamClusterWithTemplate(t testing.TB, tmplstring, clusterNamestring, numServersint) *cluster {
returncreateJetStreamClusterWithTemplateAndModHook(t, tmpl, clusterName, numServers, nil)
}

funccreateJetStreamClusterWithTemplateAndModHook(t testing.TB, tmplstring, clusterNamestring, numServersint, modifymodifyCb) *cluster {
startPorts:= []int{7_022, 9_022, 11_022, 15_022}
port:=startPorts[rand.Intn(len(startPorts))]
returncreateJetStreamClusterAndModHook(t, tmpl, clusterName, _EMPTY_, numServers, port, true, modify)
}

funccreateJetStreamCluster(t testing.TB, tmplstring, clusterName, snPrestring, numServersint, portStartint, waitOnReadybool) *cluster {
returncreateJetStreamClusterAndModHook(t, tmpl, clusterName, snPre, numServers, portStart, waitOnReady, nil)
}

typemodifyCbfunc(serverName, clusterName, storeDir, confstring) string

funccreateJetStreamClusterAndModHook(t testing.TB, tmpl, cName, snPrestring, numServersint, portStartint, waitOnReadybool, modifymodifyCb) *cluster {
returncreateJetStreamClusterEx(t, tmpl, cName, snPre, numServers, portStart, waitOnReady, modify, nil)
}

funccreateJetStreamClusterWithNetProxy(t testing.TB, cNamestring, numServersint, cnp*clusterProxy) *cluster {
startPorts:= []int{7_122, 9_122, 11_122, 15_122}
port:=startPorts[rand.Intn(len(startPorts))]
returncreateJetStreamClusterEx(t, jsClusterTempl, cName, _EMPTY_, numServers, port, true, nil, cnp)
}

funccreateJetStreamClusterEx(t testing.TB, tmpl, cName, snPrestring, numServersint, portStartint, waitbool, modifymodifyCb, cnp*clusterProxy) *cluster {
t.Helper()
ifcName==_EMPTY_||numServers<1 {
t.Fatalf("Bad params")
 }

// Flaky test prevention:
// Binding a socket to IP stack port 0 will bind an ephemeral port from an OS-specific range.
// If someone passes in to us a port spec which would cover that range, the test would be flaky.
// Adjust these ports to be the most inclusive across the port runner OSes.
// Linux: /proc/sys/net/ipv4/ip_local_port_range : 32768:60999
// <https://dataplane.org/ephemeralports.html> is useful, and shows there's no safe available range without OS-specific tuning.
// Our tests are usually run on Linux. Folks who care about other OSes: if you can't tune your test-runner OS to match, please
// propose a viable alternative.
constprohibitedPortFirst=32768
constprohibitedPortLast=60999
if (portStart>=prohibitedPortFirst&&portStart<=prohibitedPortLast) ||
 (portStart+numServers-1>=prohibitedPortFirst&&portStart+numServers-1<=prohibitedPortLast) {
t.Fatalf("test setup failure: may not specify a cluster port range which falls within %d:%d", prohibitedPortFirst, prohibitedPortLast)
 }

// Build out the routes that will be shared with all configs.
varroutes []string
varnproxies []*netProxy
forcp:=portStart; cp<portStart+numServers; cp++ {
routeURL:=fmt.Sprintf("nats-route://127.0.0.1:%d", cp)
ifcnp!=nil {
np:=createNetProxy(cnp.rtt, cnp.up, cnp.down, routeURL, false)
nproxies=append(nproxies, np)
routeURL=np.routeURL()
 }
routes=append(routes, routeURL)
 }
routeConfig:=strings.Join(routes, ",")

// Go ahead and build configurations and start servers.
c:=&cluster{servers: make([]*Server, 0, numServers), opts: make([]*Options, 0, numServers), name: cName, nproxies: nproxies}

// Start any proxies.
for_, np:=rangenproxies {
np.start()
 }

forcp:=portStart; cp<portStart+numServers; cp++ {
storeDir:=t.TempDir()
sn:=fmt.Sprintf("%sS-%d", snPre, cp-portStart+1)
conf:=fmt.Sprintf(tmpl, sn, storeDir, cName, cp, routeConfig)
ifmodify!=nil {
conf=modify(sn, cName, storeDir, conf)
 }
s, o:=RunServerWithConfig(createConfFile(t, []byte(conf)))
c.servers=append(c.servers, s)
c.opts=append(c.opts, o)
 }
c.t=t

// Wait til we are formed and have a leader.
c.checkClusterFormed()
ifwait {
c.waitOnClusterReady()
 }

returnc
}

func (c*cluster) addInNewServer() *Server {
c.t.Helper()
sn:=fmt.Sprintf("S-%d", len(c.servers)+1)
storeDir:=c.t.TempDir()
seedRoute:=fmt.Sprintf("nats-route://127.0.0.1:%d", c.opts[0].Cluster.Port)
conf:=fmt.Sprintf(jsClusterTempl, sn, storeDir, c.name, -1, seedRoute)
s, o:=RunServerWithConfig(createConfFile(c.t, []byte(conf)))
c.servers=append(c.servers, s)
c.opts=append(c.opts, o)
c.checkClusterFormed()
returns
}

// This is tied to jsClusterAccountsTempl, so changes there to users needs to be reflected here.
func (c*cluster) createSingleLeafNodeNoSystemAccount() *Server {
as:=c.randomServer()
lno:=as.getOpts().LeafNode
ln1:=fmt.Sprintf("nats://one:p@%s:%d", lno.Host, lno.Port)
ln2:=fmt.Sprintf("nats://two:p@%s:%d", lno.Host, lno.Port)
conf:=fmt.Sprintf(jsClusterSingleLeafNodeTempl, c.t.TempDir(), ln1, ln2)
s, o:=RunServerWithConfig(createConfFile(c.t, []byte(conf)))
c.servers=append(c.servers, s)
c.opts=append(c.opts, o)

checkLeafNodeConnectedCount(c.t, as, 2)

returns
}

// This is tied to jsClusterAccountsTempl, so changes there to users needs to be reflected here.
func (c*cluster) createSingleLeafNodeNoSystemAccountAndEnablesJetStream() *Server {
returnc.createSingleLeafNodeNoSystemAccountAndEnablesJetStreamWithDomain(_EMPTY_, "nojs")
}

func (c*cluster) createSingleLeafNodeNoSystemAccountAndEnablesJetStreamWithDomain(domain, userstring) *Server {
tmpl:=jsClusterSingleLeafNodeLikeNGSTempl
ifdomain!=_EMPTY_ {
nsc:=fmt.Sprintf("domain: %s, store_dir:", domain)
tmpl=strings.Replace(jsClusterSingleLeafNodeLikeNGSTempl, "store_dir:", nsc, 1)
 }
as:=c.randomServer()
lno:=as.getOpts().LeafNode
ln:=fmt.Sprintf("nats://%s:p@%s:%d", user, lno.Host, lno.Port)
conf:=fmt.Sprintf(tmpl, c.t.TempDir(), ln)
s, o:=RunServerWithConfig(createConfFile(c.t, []byte(conf)))
c.servers=append(c.servers, s)
c.opts=append(c.opts, o)

checkLeafNodeConnectedCount(c.t, as, 1)

returns
}

varjsClusterSingleLeafNodeLikeNGSTempl=`
 listen: 127.0.0.1:-1
 server_name: LNJS
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 leaf { remotes [ { urls: [ %s ] } ] }
`

varjsClusterSingleLeafNodeTempl=`
 listen: 127.0.0.1:-1
 server_name: LNJS
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 leaf { remotes [
 { urls: [ %s ], account: "JSY" }
 { urls: [ %s ], account: "JSN" } ]
 }

 accounts {
 JSY { users = [ { user: "y", pass: "p" } ]; jetstream: true }
 JSN { users = [ { user: "n", pass: "p" } ] }
 $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
 }
`

varjsClusterTemplWithLeafNode=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 {{leaf}}

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 # For access to system account.
 accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
`

varjsClusterTemplWithLeafNodeNoJS=`
 listen: 127.0.0.1:-1
 server_name: %s

 # Need to keep below since it fills in the store dir by default so just comment out.
 # jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 {{leaf}}

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 # For access to system account.
 accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
`

varjsClusterTemplWithSingleLeafNode=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 {{leaf}}

 # For access to system account.
 accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
`

varjsClusterTemplWithSingleFleetLeafNode=`
 listen: 127.0.0.1:-1
 server_name: %s
 cluster: { name: fleet }
 jetstream: {max_mem_store: 2GB, max_file_store: 2GB, store_dir: '%s'}

 {{leaf}}

 # For access to system account.
 accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
`

varjsClusterTemplWithSingleLeafNodeNoJS=`
 listen: 127.0.0.1:-1
 server_name: %s

 # jetstream: {store_dir: '%s'}

 {{leaf}}