jetstream_cluster_long_test.go

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

// This test file is skipped by default to avoid accidentally running (e.g. `go test ./server`)
//go:build !skip_js_tests && include_js_long_tests

package server

import (
"context"
"errors"
"fmt"
"math/rand"
"slices"
"strings"
"sync"
"sync/atomic"
"testing"
"time"

"github.com/nats-io/nats.go"
"github.com/nats-io/nuid"
)

// TestLongKVPutWithServerRestarts overwrites values in a replicated KV bucket for a fixed amount of time.
// Also restarts a random server at fixed interval.
// The test fails if updates fail for a continuous interval of time, or if a server fails to restart and catch up.
funcTestLongKVPutWithServerRestarts(t*testing.T) {
// RNG Seed
constSeed=123456
// Number of keys in bucket
constNumKeys=1000
// Size of (random) values
constValueSize=1024
// Test duration
constDuration=3*time.Minute
// If no updates successful for this interval, then fail the test
constMaxRetry=5*time.Second
// Minimum time between put operations
constUpdatesInterval=1*time.Millisecond
// Time between progress reports to console
constProgressInterval=10*time.Second
// Time between server restarts
constServerRestartInterval=5*time.Second

typeParametersstruct {
numServersint
replicasint
storage nats.StorageType
 }

test:=func(t*testing.T, pParameters) {
rng:=rand.New(rand.NewSource(Seed))

// Create cluster
clusterName:=fmt.Sprintf("C_%d-%s", p.numServers, p.storage)
cluster:=createJetStreamClusterExplicit(t, clusterName, p.numServers)
defercluster.shutdown()

// Connect to a random server but client will discover others too.
nc, js:=jsClientConnect(t, cluster.randomServer())
defernc.Close()

// Create bucket
kv, err:=js.CreateKeyValue(&nats.KeyValueConfig{
Bucket: "TEST",
Replicas: p.replicas,
Storage: p.storage,
 })
require_NoError(t, err)

// Initialize list of keys
keys:=make([]string, NumKeys)
fori:=0; i<NumKeys; i++ {
keys[i] =fmt.Sprintf("key-%d", i)
 }

// Initialize keys in bucket with an empty value
for_, key:=rangekeys {
_, err:=kv.Put(key, []byte{})
require_NoError(t, err)
 }

// Track statistics
varstats=struct {
start time.Time
lastSuccessfulUpdate time.Time
updateOkuint64
updateErruint64
restartsuint64
restartsMapmap[string]int
 }{
start: time.Now(),
lastSuccessfulUpdate: time.Now(),
restartsMap: make(map[string]int, p.numServers),
 }
for_, server:=rangecluster.servers {
stats.restartsMap[server.Name()] =0
 }

// Print statistics
printProgress:=func() {

t.Logf(
"[%s] %d updates %d errors, %d server restarts (%v)",
time.Since(stats.start).Round(time.Second),
stats.updateOk,
stats.updateErr,
stats.restarts,
stats.restartsMap,
 )
 }

// Print update on completion
deferprintProgress()

// Pick a random key and update it with a random value
valueBuffer:=make([]byte, ValueSize)
updateRandomKey:=func() error {
key:=keys[rand.Intn(NumKeys)]
_, err:=rng.Read(valueBuffer)
require_NoError(t, err)
_, err=kv.Put(key, valueBuffer)
returnerr
 }

// Set up timers and tickers
endTestTimer:=time.After(Duration)
nextUpdateTicker:=time.NewTicker(UpdatesInterval)
progressTicker:=time.NewTicker(ProgressInterval)
restartServerTicker:=time.NewTicker(ServerRestartInterval)

 runLoop:
for {
select {

case<-endTestTimer:
break runLoop

case<-nextUpdateTicker.C:
err:=updateRandomKey()
iferr==nil {
stats.updateOk++
stats.lastSuccessfulUpdate=time.Now()
 } else {
stats.updateErr++
iftime.Since(stats.lastSuccessfulUpdate) >MaxRetry {
t.Fatalf("Could not successfully update for over %s", MaxRetry)
 }
 }

case<-restartServerTicker.C:
randomServer:=cluster.servers[rng.Intn(len(cluster.servers))]
randomServer.Shutdown()
randomServer.WaitForShutdown()
restartedServer:=cluster.restartServer(randomServer)
cluster.waitOnClusterReady()
cluster.waitOnServerHealthz(restartedServer)
cluster.waitOnAllCurrent()
stats.restarts++
stats.restartsMap[randomServer.Name()]++

case<-progressTicker.C:
printProgress()
 }
 }
 }

testCases:= []Parameters{
 {numServers: 5, replicas: 3, storage: nats.MemoryStorage},
 {numServers: 5, replicas: 3, storage: nats.FileStorage},
 }

for_, testCase:=rangetestCases {
name:=fmt.Sprintf("N:%d,R:%d,%s", testCase.numServers, testCase.replicas, testCase.storage)
t.Run(
name,
func(t*testing.T) { test(t, testCase) },
 )
 }
}

// This is a RaftChainOfBlocks test that randomly starts and stops nodes to exercise recovery and snapshots.
funcTestLongNRGChainOfBlocks(t*testing.T) {
const (
ClusterSize=3
GroupSize=3
ConvergenceTimeout=30*time.Second
Duration=10*time.Minute
PrintStateInterval=3*time.Second
 )

// Create cluster
c:=createJetStreamClusterExplicit(t, "Test", ClusterSize)
deferc.shutdown()

rg:=c.createRaftGroup("ChainOfBlocks", GroupSize, newRaftChainStateMachine)
rg.waitOnLeader()

// Available operations
typeTestOperationstring
const (
StopOneTestOperation="Stop one active node"
StopAll="Stop all active nodes"
RestartOne="Restart one stopped node"
RestartAll="Restart all stopped nodes"
Snapshot="Snapshot one active node"
Propose="Propose a value via one active node"
ProposeLeader="Propose a value via leader"
Pause="Let things run undisturbed for a while"
Check="Wait for nodes to converge"
 )

// Weighted distribution of operations, one is randomly chosen from this vector in each iteration
opsWeighted:= []TestOperation{
StopOne,
StopOne,
StopOne,
StopAll,
StopAll,
RestartOne,
RestartOne,
RestartOne,
RestartAll,
RestartAll,
RestartAll,
Snapshot,
Snapshot,
Propose,
Propose,
Propose,
Propose,
ProposeLeader,
ProposeLeader,
ProposeLeader,
ProposeLeader,
Pause,
Pause,
Pause,
Pause,
Check,
Check,
Check,
Check,
 }

rngSeed:=time.Now().UnixNano()
rng:=rand.New(rand.NewSource(rngSeed))
t.Logf("Seed: %d", rngSeed)

// Chose a node from the list (and remove it)
pickRandomNode:=func(nodes []stateMachine) ([]stateMachine, stateMachine) {
iflen(nodes) ==0 {
// Input list is empty
returnnodes, nil
 }
// Pick random node
i:=rng.Intn(len(nodes))
node:=nodes[i]
// Move last element in its place
nodes[i] =nodes[len(nodes)-1]
// Return slice excluding last element
returnnodes[:len(nodes)-1], node
 }

// Create summary status string for all replicas
chainStatusString:=func() string {
b:= strings.Builder{}
for_, sm:=rangerg {
csm:=sm.(*RCOBStateMachine)
running, blocksCount, blockHash:=csm.getCurrentHash()
ifrunning {
b.WriteString(
fmt.Sprintf(
" [%s (%s): %d blocks, hash=%s],",
csm.server().Name(),
csm.node().ID(),
blocksCount,
blockHash,
 ),
 )
 } else {
b.WriteString(
fmt.Sprintf(
" [%s (%s): STOPPED],",
csm.server().Name(),
csm.node().ID(),
 ),
 )

 }
 }
returnb.String()
 }

// Track the highest number of blocks applied by any of the replicas
highestBlocksCount:=uint64(0)

// Track active and stopped nodes
activeNodes:=make([]stateMachine, 0, GroupSize)
stoppedNodes:=make([]stateMachine, 0, GroupSize)

// Initially all nodes are active
activeNodes=append(activeNodes, rg...)

deferfunc() {
t.Logf("Final state: %s", chainStatusString())
 }()

printStateTicker:=time.NewTicker(PrintStateInterval)
testTimer:=time.NewTimer(Duration)
start:=time.Now()
iteration:=0
stopCooldown:=0

for {

iteration++
select {
case<-printStateTicker.C:
t.Logf(
"[%s] State: %s",
time.Since(start).Round(time.Second),
chainStatusString(),
 )
case<-testTimer.C:
// Test completed
return
default:
// Continue
 }

// Choose a random operation to perform in this iteration
nextOperation:=opsWeighted[rng.Intn(len(opsWeighted))]

// If we're about to stop one or all nodes, do some sanity checks to ensure we don't
// spam stops which would constantly interrupt the chance of making progress.
ifnextOperation==StopOne||nextOperation==StopAll {
ifstopCooldown>0 {
nextOperation=Propose
 } else {
stopCooldown=50
 }
 }
ifstopCooldown>0 {
stopCooldown--
 }

ifRCOBOptions.verbose {
t.Logf("Iteration %d: %s", iteration, nextOperation)
 }

switchnextOperation {

caseStopOne:
// Stop an active node (if any are left active)
varnstateMachine
activeNodes, n=pickRandomNode(activeNodes)
ifn!=nil {
n.stop()
stoppedNodes=append(stoppedNodes, n)
 }

caseStopAll:
// Stop all active nodes (if any are active)
for_, node:=rangeactiveNodes {
node.stop()
 }
stoppedNodes=append(stoppedNodes, activeNodes...)
activeNodes=make([]stateMachine, 0, GroupSize)

caseRestartOne:
// Restart a stopped node (if any are stopped)
varnstateMachine
stoppedNodes, n=pickRandomNode(stoppedNodes)
ifn!=nil {
n.restart()
activeNodes=append(activeNodes, n)
 }

caseRestartAll:
// Restart all stopped nodes (if any)
for_, node:=rangestoppedNodes {
node.restart()
 }
activeNodes=append(activeNodes, stoppedNodes...)
stoppedNodes=make([]stateMachine, 0, GroupSize)

caseSnapshot:
// Choose a random active node and tell it to create a snapshot
iflen(activeNodes) >0 {
n:=activeNodes[rng.Intn(len(activeNodes))]
n.(*RCOBStateMachine).createSnapshot()
 }

casePropose:
// Make an active node propose the next block (if any nodes are active)
iflen(activeNodes) >0 {
n:=activeNodes[rng.Intn(len(activeNodes))]
n.(*RCOBStateMachine).proposeBlock()
 }

caseProposeLeader:
// Make the leader propose the next block (if a leader is active)
leader:=rg.leader()
ifleader!=nil {
leader.(*RCOBStateMachine).proposeBlock()
 }

casePause:
// Noop, let things run undisturbed for a little bit
time.Sleep(time.Duration(rng.Intn(250)) *time.Millisecond)

caseCheck:
// Restart any stopped node
for_, node:=rangestoppedNodes {
node.restart()
 }
activeNodes=append(activeNodes, stoppedNodes...)
stoppedNodes=make([]stateMachine, 0, GroupSize)

// Ensure all nodes (eventually) converge
checkFor(t, ConvergenceTimeout, 250*time.Millisecond,
func() error {
referenceNode:=rg[0]
// Save block count and hash of first node as reference
_, referenceBlocksCount, referenceHash:=referenceNode.(*RCOBStateMachine).getCurrentHash()

// Compare each node against reference
for_, n:=rangerg {
sm:=n.(*RCOBStateMachine)
running, blocksCount, blockHash:=sm.getCurrentHash()
if!running {
returnfmt.Errorf(
"node not running: %s (%s)",
sm.server().Name(),
sm.node().ID(),
 )
 }

// Track the highest block delivered by any node
ifblocksCount>highestBlocksCount {
ifRCOBOptions.verbose {
t.Logf(
"New highest blocks count: %d (%s (%s))",
blocksCount,
sm.s.Name(),
sm.n.ID(),
 )
 }
highestBlocksCount=blocksCount
 }

// Each replica must match the reference node

ifblocksCount!=referenceBlocksCount {
returnfmt.Errorf(
"different number of blocks %d (%s (%s) vs. %d (%s (%s))",
blocksCount,
sm.server().Name(),
sm.node().ID(),
referenceBlocksCount,
referenceNode.server().Name(),
referenceNode.node().ID(),
 )
 } elseifblockHash!=referenceHash {
returnfmt.Errorf(
"different hash after %d blocks %s (%s (%s) vs. %s (%s (%s))",
blocksCount,
blockHash,
sm.server().Name(),
sm.node().ID(),
referenceHash,
referenceNode.server().Name(),
referenceNode.node().ID(),
 )
 }
 }

// Verify consistency check was against the highest block known
ifreferenceBlocksCount<highestBlocksCount {
returnfmt.Errorf(
"nodes converged below highest known block count: %d: %s",
highestBlocksCount,
chainStatusString(),
 )
 }

// All nodes reached the same state, check passed
returnnil
 },
 )
 }
 }
}

funcTestLongClusterWorkQueueMessagesNotSkipped(t*testing.T) {
c:=createJetStreamClusterExplicit(t, "R3S", 3)
deferc.shutdown()

nc, js:=jsClientConnect(t, c.randomServer())
defernc.Close()

iterations:=500_000
stream:="s1"
subjf:="subj.>"
consumers:=map[string]string{
"c1": "subj.c1",
"c2": "subj.c2.*",
"c3": "subj.c3",
 }

sig:=make(chan*nats.Msg, 900)

_, err:=js.AddStream(&nats.StreamConfig{
Name: stream,
Storage: nats.FileStorage,
Subjects: []string{subjf},
Retention: nats.WorkQueuePolicy,
Duplicates: time.Minute*2,
Replicas: 3,
MaxAge: time.Hour,
 })
require_NoError(t, err)

forname, subjf:=rangeconsumers {
nc, js:=jsClientConnect(t, c.randomServer())
defernc.Close()

_, err=js.AddConsumer(stream, &nats.ConsumerConfig{
Name: name,
FilterSubject: subjf,
Replicas: 3,
AckPolicy: nats.AckExplicitPolicy,
DeliverPolicy: nats.DeliverAllPolicy,
 })
require_NoError(t, err)

ps, err:=js.PullSubscribe(subjf, "", nats.Bind(stream, name))
require_NoError(t, err)

gofunc() {
for {
msgs, err:=ps.FetchBatch(300)
iferrors.Is(err, nats.ErrTimeout) {
continue
 }
iferrors.Is(err, nats.ErrConnectionClosed) ||errors.Is(err, nats.ErrSubscriptionClosed) {
return// ... for when the test finishes
 }
require_NoError(t, err)
formsg:=rangemsgs.Messages() {
gofunc() {
time.Sleep(time.Millisecond*time.Duration(rand.Int31n(100)))
require_NoError(t, msg.Ack())
sig<-msg
 }()
 }
 }
 }()
 }

gofunc() {
hdrs:= nats.Header{}
fori:=1; i<=iterations; i++ {
// Pick a random consumer to hit this time (map iteration order is
// non-deterministic, but break to do it just once).
for_, subj:=rangeconsumers {
hdrs.Set("Nats-Msg-Id", fmt.Sprintf("msg-%d", i))
ifstrings.HasPrefix(subj, "*") {
subj=strings.Replace(subj, "*", fmt.Sprintf("%d", i), 1)
 }
_, err:=js.PublishMsg(&nats.Msg{
Subject: subj,
Header: hdrs,
 })
require_NoError(t, err)
break
 }
 }
 }()

fori:=1; i<=iterations; i++ {
ifi%10000==0 {
t.Logf("%d messages out of %d", i, iterations)
 }

select {
case<-sig:
case<-time.After(time.Second*2):
si, err:=js.StreamInfo(stream)
require_NoError(t, err)
t.Logf("Stream info: %+v", si.State)

forname:=rangeconsumers {
ci, err:=js.ConsumerInfo(stream, name)
require_NoError(t, err)
t.Logf("Consumer %q info: %+v, %+v", name, ci.AckFloor, ci.Delivered)
 }

t.Fatalf("Didn't receive message %d", i)
 }
 }
}

funcTestLongClusterStreamOrphanMsgsAndReplicasDrifting(t*testing.T) {
checkInterestStateT=4*time.Second
checkInterestStateJ=1
deferfunc() {
checkInterestStateT=defaultCheckInterestStateT
checkInterestStateJ=defaultCheckInterestStateJ
 }()

typetestParamsstruct {
restartAnybool
restartLeaderbool
rolloutRestartbool
ldmRestartbool
restartsint
checkHealthzbool
reconnectRoutesbool
reconnectClientsbool
 }
test:=func(t*testing.T, params*testParams, sc*nats.StreamConfig) {
conf:=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {
 store_dir: '%s',
 }
 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }
 server_tags: ["test"]
 system_account: sys
 no_auth_user: js
 accounts {
 sys { users = [ { user: sys, pass: sys } ] }
 js {
 jetstream = enabled
 users = [ { user: js, pass: js } ]
 }
 }`
c:=createJetStreamClusterWithTemplate(t, conf, sc.Name, 3)
deferc.shutdown()

// Update lame duck duration for all servers.
for_, s:=rangec.servers {
s.optsMu.Lock()
s.opts.LameDuckDuration=5*time.Second
s.opts.LameDuckGracePeriod=-5*time.Second
s.optsMu.Unlock()
 }

nc, js:=jsClientConnect(t, c.randomServer())
defernc.Close()

cnc, cjs:=jsClientConnect(t, c.randomServer())
defercnc.Close()

_, err:=js.AddStream(sc)
require_NoError(t, err)

pctx, cancel:=context.WithTimeout(context.Background(), 60*time.Second)
defercancel()

// Start producers
varwg sync.WaitGroup

// First call is just to create the pull subscribers.
mp:=nats.MaxAckPending(10000)
mw:=nats.PullMaxWaiting(1000)
aw:=nats.AckWait(5*time.Second)

fori:=0; i<10; i++ {
for_, partition:=range []string{"EEEEE"} {
subject:=fmt.Sprintf("MSGS.%s.*.H.100XY.*.*.WQ.00000000000%d", partition, i)
consumer:=fmt.Sprintf("consumer:%s:%d", partition, i)
_, err:=cjs.PullSubscribe(subject, consumer, mp, mw, aw)
require_NoError(t, err)
 }
 }

// Create a single consumer that does no activity.
// Make sure we still calculate low ack properly and cleanup etc.
_, err=cjs.PullSubscribe("MSGS.ZZ.>", "consumer:ZZ:0", mp, mw, aw)
require_NoError(t, err)

subjects:= []string{
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000000",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000001",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000002",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000003",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000004",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000005",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000006",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000007",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000008",
"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000009",
 }
payload:= []byte(strings.Repeat("A", 1024))

fori:=0; i<50; i++ {
wg.Add(1)
gofunc() {
pnc, pjs:=jsClientConnect(t, c.randomServer())
deferpnc.Close()

fori:=1; i<200_000; i++ {
select {
case<-pctx.Done():
wg.Done()
return
default:
 }
for_, subject:=rangesubjects {
// Send each message a few times.
msgID:=nats.MsgId(nuid.Next())
pjs.PublishAsync(subject, payload, msgID)
pjs.Publish(subject, payload, msgID, nats.AckWait(250*time.Millisecond))
pjs.Publish(subject, payload, msgID, nats.AckWait(250*time.Millisecond))
 }
 }
 }()
 }

// Rogue publisher that sends the same msg ID everytime.
fori:=0; i<10; i++ {
wg.Add(1)
gofunc() {
pnc, pjs:=jsClientConnect(t, c.randomServer())
deferpnc.Close()

msgID:=nats.MsgId("1234567890")
fori:=1; ; i++ {
select {
case<-pctx.Done():
wg.Done()
return
default:
 }
for_, subject:=rangesubjects {
// Send each message a few times.
pjs.PublishAsync(subject, payload, msgID, nats.RetryAttempts(0), nats.RetryWait(0))
pjs.Publish(subject, payload, msgID, nats.AckWait(1*time.Millisecond), nats.RetryAttempts(0), nats.RetryWait(0))
pjs.Publish(subject, payload, msgID, nats.AckWait(1*time.Millisecond), nats.RetryAttempts(0), nats.RetryWait(0))
 }
 }
 }()
 }

// Let enough messages into the stream then start consumers.
time.Sleep(15*time.Second)

ctx, cancel:=context.WithTimeout(context.Background(), 45*time.Second)
defercancel()

fori:=0; i<10; i++ {
subject:=fmt.Sprintf("MSGS.EEEEE.*.H.100XY.*.*.WQ.00000000000%d", i)
consumer:=fmt.Sprintf("consumer:EEEEE:%d", i)
forn:=0; n<5; n++ {
cpnc, cpjs:=jsClientConnect(t, c.randomServer())
defercpnc.Close()

psub, err:=cpjs.PullSubscribe(subject, consumer, mp, mw, aw)
require_NoError(t, err)

time.AfterFunc(15*time.Second, func() {
cpnc.Close()
 })

wg.Add(1)
gofunc() {
tick:=time.NewTicker(1*time.Millisecond)
for {
ifcpnc.IsClosed() {
wg.Done()
return
 }
select {
case<-ctx.Done():
wg.Done()
return
case<-tick.C:
// Fetch 1 first, then if no errors Fetch 100.
msgs, err:=psub.Fetch(1, nats.MaxWait(200*time.Millisecond))
iferr!=nil {
continue
 }
for_, msg:=rangemsgs {
msg.Ack()
 }
msgs, err=psub.Fetch(100, nats.MaxWait(200*time.Millisecond))
iferr!=nil {
continue
 }
for_, msg:=rangemsgs {
msg.Ack()
 }
msgs, err=psub.Fetch(1000, nats.MaxWait(200*time.Millisecond))
iferr!=nil {
continue
 }
for_, msg:=rangemsgs {
msg.Ack()
 }
 }
 }
 }()
 }
 }

fori:=0; i<10; i++ {
subject:=fmt.Sprintf("MSGS.EEEEE.*.H.100XY.*.*.WQ.00000000000%d", i)
consumer:=fmt.Sprintf("consumer:EEEEE:%d", i)
forn:=0; n<10; n++ {
cpnc, cpjs:=jsClientConnect(t, c.randomServer())
defercpnc.Close()

psub, err:=cpjs.PullSubscribe(subject, consumer, mp, mw, aw)
iferr!=nil {
t.Logf("ERROR: %v", err)
continue
 }

wg.Add(1)
gofunc() {
tick:=time.NewTicker(1*time.Millisecond)
for {
select {
case<-ctx.Done():
wg.Done()
return
case<-tick.C:
// Fetch 1 first, then if no errors Fetch 100.
msgs, err:=psub.Fetch(1, nats.MaxWait(200*time.Millisecond))
iferr!=nil {
continue
 }
for_, msg:=rangemsgs {
msg.Ack()
 }
msgs, err=psub.Fetch(100, nats.MaxWait(200*time.Millisecond))
iferr!=nil {
continue
 }
for_, msg:=rangemsgs {
msg.Ack()
 }

msgs, err=psub.Fetch(1000, nats.MaxWait(200*time.Millisecond))
iferr!=nil {
continue
 }
for_, msg:=rangemsgs {
msg.Ack()
 }
 }
 }
 }()
 }
 }

// Periodically disconnect routes from one of the servers.
ifparams.reconnectRoutes {
wg.Add(1)
gofunc() {
forrangetime.NewTicker(10*time.Second).C {
select {
case<-ctx.Done():
wg.Done()
return
default:
 }

// Force disconnecting routes from one of the servers.
s:=c.servers[rand.Intn(3)]
varroutes []*client
t.Logf("Disconnecting routes from %v", s.Name())
s.mu.Lock()
for_, conns:=ranges.routes {
routes=append(routes, conns...)
 }
s.mu.Unlock()
for_, r:=rangeroutes {
r.closeConnection(ClientClosed)
 }
 }
 }()
 }

// Periodically reconnect clients.
ifparams.reconnectClients {
reconnectClients:=func(s*Server) {
for_, client:=ranges.clients {
client.closeConnection(Kicked)
 }
 }

wg.Add(1)
gofunc() {
forrangetime.NewTicker(10*time.Second).C {
select {
case<-ctx.Done():
wg.Done()
return
default:
 }
// Force reconnect clients from one of the servers.
s:=c.servers[rand.Intn(len(c.servers))]
reconnectClients(s)
 }
 }()
 }

// Restarts
wg.Add(1)
time.AfterFunc(10*time.Second, func() {
deferwg.Done()
fori:=0; i<params.restarts; i++ {
select {
case<-ctx.Done():
return
default:
// Keep going
 }

switch {
caseparams.restartLeader:
// Find server leader of the stream and restart it.
s:=c.streamLeader("js", sc.Name)
ifparams.ldmRestart {
s.lameDuckMode()
 } else {
s.Shutdown()
 }
s.WaitForShutdown()
c.restartServer(s)
caseparams.restartAny:
s:=c.servers[rand.Intn(len(c.servers))]
ifparams.ldmRestart {
s.lameDuckMode()
 } else {
s.Shutdown()
 }
s.WaitForShutdown()
c.restartServer(s)
caseparams.rolloutRestart:
for_, s:=rangec.servers {
ifparams.ldmRestart {
s.lameDuckMode()
 } else {
s.Shutdown()
 }
s.WaitForShutdown()
c.restartServer(s)

ifparams.checkHealthz {
hctx, hcancel:=context.WithTimeout(ctx, 15*time.Second)
deferhcancel()

forrangetime.NewTicker(2*time.Second).C {
select {
case<-hctx.Done():
default:
 }

status:=s.healthz(nil)
ifstatus.StatusCode==200 {
break
 }
 }
 }
 }
 }
c.waitOnClusterReady()
 }
 })

// Wait until context is done then check state.
<-ctx.Done()

getStreamDetails:=func(t*testing.T, srv*Server) *StreamDetail {
t.Helper()
jsz, err:=srv.Jsz(&JSzOptions{Accounts: true, Streams: true, Consumer: true})
require_NoError(t, err)
iflen(jsz.AccountDetails) >0&&len(jsz.AccountDetails[0].Streams) >0 {
stream:=jsz.AccountDetails[0].Streams[0]