jetstream_cluster_4_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.

//go:build !skip_js_tests && !skip_js_cluster_tests_4

package server

import (
"context"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"math/rand"
"os"
"path"
"path/filepath"
"runtime"
"slices"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"

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

funcTestJetStreamClusterWorkQueueStreamDiscardNewDesync(t*testing.T) {
t.Run("max msgs", func(t*testing.T) {
testJetStreamClusterWorkQueueStreamDiscardNewDesync(t, &nats.StreamConfig{
Name: "WQTEST_MM",
Subjects: []string{"messages.*"},
Replicas: 3,
MaxAge: 10*time.Minute,
MaxMsgs: 100,
Retention: nats.WorkQueuePolicy,
Discard: nats.DiscardNew,
 })
 })
t.Run("max bytes", func(t*testing.T) {
testJetStreamClusterWorkQueueStreamDiscardNewDesync(t, &nats.StreamConfig{
Name: "WQTEST_MB",
Subjects: []string{"messages.*"},
Replicas: 3,
MaxAge: 10*time.Minute,
MaxBytes: 1*1024*1024,
Retention: nats.WorkQueuePolicy,
Discard: nats.DiscardNew,
 })
 })
}

functestJetStreamClusterWorkQueueStreamDiscardNewDesync(t*testing.T, 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]
 }
 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()

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

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

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

ctx, cancel:=context.WithCancel(context.Background())
defercancel()

psub, err:=cjs.PullSubscribe("messages.*", "consumer")
require_NoError(t, err)

stepDown:=func() {
_, err=nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, sc.Name), nil, time.Second)
 }

// Messages will be produced and consumed in parallel, then once there are
// enough errors a leader election will be triggered.
var (
wg sync.WaitGroup
receiveduint64
errCh=make(chanerror, 100_000)
receivedMap=make(map[string]*nats.Msg)
 )
wg.Add(1)
gofunc() {
tick:=time.NewTicker(20*time.Millisecond)
for {
select {
case<-ctx.Done():
wg.Done()
return
case<-tick.C:
msgs, err:=psub.Fetch(10, nats.MaxWait(200*time.Millisecond))
iferr!=nil {
// The consumer will continue to timeout here eventually.
continue
 }
for_, msg:=rangemsgs {
received++
receivedMap[msg.Subject] =msg
msg.Ack()
 }
 }
 }
 }()

shouldDrop:=make(map[string]error)
wg.Add(1)
gofunc() {
payload:= []byte(strings.Repeat("A", 1024))
tick:=time.NewTicker(1*time.Millisecond)
fori:=1; ; i++ {
select {
case<-ctx.Done():
wg.Done()
return
case<-tick.C:
subject:=fmt.Sprintf("messages.%d", i)
_, err:=js.Publish(subject, payload, nats.RetryAttempts(0))
// Capture the messages that have failed.
iferr!=nil {
// Unless it was a timeout, we can't be sure if the message was received,
// and we just didn't get a response.
iferrors.Is(err, nats.ErrTimeout) {
break
 }
errCh<-err
shouldDrop[subject] =err
 }
 }
 }
 }()

// Collect enough errors to cause things to get out of sync.
varerrCountint
Setup:
for {
select {
caseerr=<-errCh:
errCount++
iferrCount%500==0 {
stepDown()
 } elseiferrCount>=2000 {
// Stop both producing and consuming.
cancel()
break Setup
 }
case<-time.After(5*time.Second):
// Unblock the test and continue.
cancel()
break Setup
 }
 }

// Both goroutines should be exiting now..
wg.Wait()

// Let acks propagate for stream checks.
time.Sleep(250*time.Millisecond)

// Check messages that ought to have been dropped.
forsubject:=rangereceivedMap {
found, ok:=shouldDrop[subject]
ifok {
t.Errorf("Should have dropped message published on %q since got error: %v", subject, found)
 }
 }
}

// https://github.com/nats-io/nats-server/issues/5071
funcTestJetStreamClusterStreamPlacementDistribution(t*testing.T) {
c:=createJetStreamClusterExplicit(t, "R3S", 5)
deferc.shutdown()

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

fori:=1; i<=10; i++ {
_, err:=js.AddStream(&nats.StreamConfig{
Name: fmt.Sprintf("TEST:%d", i),
Subjects: []string{fmt.Sprintf("foo.%d.*", i)},
Replicas: 3,
 })
require_NoError(t, err)
 }

// 10 streams, 3 replicas div 5 servers.
expectedStreams:=10*3/5
for_, s:=rangec.servers {
jsz, err:=s.Jsz(nil)
require_NoError(t, err)
require_Equal(t, jsz.Streams, expectedStreams)
 }
}

funcTestJetStreamClusterSourceWorkingQueueWithLimit(t*testing.T) {
const (
totalMsgs=300
maxMsgs=100
maxBytes=maxMsgs*100
msgPayloadFormat="%0100d"// %0100d is 100 bytes. Must match payload value above.
 )
c:=createJetStreamClusterExplicit(t, "WQ3", 3)
deferc.shutdown()

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

_, err:=js.AddStream(&nats.StreamConfig{Name: "test", Subjects: []string{"test"}, Replicas: 3})
require_NoError(t, err)

_, err=js.AddStream(&nats.StreamConfig{Name: "wq", MaxMsgs: maxMsgs, Discard: nats.DiscardNew, Retention: nats.WorkQueuePolicy,
Sources: []*nats.StreamSource{{Name: "test"}}, Replicas: 3})
require_NoError(t, err)

_, err=js.AddStream(&nats.StreamConfig{Name: "wq2", MaxBytes: maxBytes, Discard: nats.DiscardNew, Retention: nats.WorkQueuePolicy,
Sources: []*nats.StreamSource{{Name: "test"}}, Replicas: 3})
require_NoError(t, err)

sendBatch:=func(subjectstring, nint) {
fori:=0; i<n; i++ {
_, err=js.Publish(subject, []byte(fmt.Sprintf(msgPayloadFormat, i)))
require_NoError(t, err)
 }
 }

f:=func(ss*nats.Subscription, donechanbool) {
fori:=0; i<totalMsgs; i++ {
m, err:=ss.Fetch(1, nats.MaxWait(3*time.Second))
require_NoError(t, err)
p, err:=strconv.Atoi(string(m[0].Data))
require_NoError(t, err)
require_Equal(t, p, i)
time.Sleep(11*time.Millisecond)
err=m[0].Ack()
require_NoError(t, err)
 }
done<-true
 }

// Populate the sourced stream.
sendBatch("test", totalMsgs)

checkFor(t, 3*time.Second, 250*time.Millisecond, func() error {
si, err:=js.StreamInfo("wq")
require_NoError(t, err)
ifsi.State.Msgs!=maxMsgs {
returnfmt.Errorf("expected %d msgs on stream wq, got state: %+v", maxMsgs, si.State)
 }
returnnil
 })

checkFor(t, 3*time.Second, 250*time.Millisecond, func() error {
si, err:=js.StreamInfo("wq2")
require_NoError(t, err)
ifsi.State.Bytes>maxBytes {
returnfmt.Errorf("expected no more than %d bytes on stream wq2, got state: %+v", maxBytes, si.State)
 }
returnnil
 })

_, err=js.AddConsumer("wq", &nats.ConsumerConfig{Durable: "wqc", FilterSubject: "test", AckPolicy: nats.AckExplicitPolicy})
require_NoError(t, err)

ss1, err:=js.PullSubscribe("test", "wqc", nats.Bind("wq", "wqc"))
require_NoError(t, err)

vardoneChan1=make(chanbool)
gof(ss1, doneChan1)

checkFor(t, 10*time.Second, 250*time.Millisecond, func() error {
si, err:=js.StreamInfo("wq")
require_NoError(t, err)
ifsi.State.Msgs>0&&si.State.Msgs<=maxMsgs {
returnfmt.Errorf("expected 0 msgs on stream wq, got: %d", si.State.Msgs)
 } elseifsi.State.Msgs>maxMsgs {
t.Fatalf("got more than our %d message limit on stream wq: %+v", maxMsgs, si.State)
 }

returnnil
 })

select {
case<-doneChan1:
ss1.Drain()
case<-time.After(10*time.Second):
t.Fatalf("Did not receive completion signal")
 }

_, err=js.AddConsumer("wq2", &nats.ConsumerConfig{Durable: "wqc", FilterSubject: "test", AckPolicy: nats.AckExplicitPolicy})
require_NoError(t, err)

ss2, err:=js.PullSubscribe("test", "wqc", nats.Bind("wq2", "wqc"))
require_NoError(t, err)

vardoneChan2=make(chanbool)
gof(ss2, doneChan2)

checkFor(t, 10*time.Second, 250*time.Millisecond, func() error {
si, err:=js.StreamInfo("wq2")
require_NoError(t, err)
ifsi.State.Bytes>0&&si.State.Bytes<=maxBytes {
returnfmt.Errorf("expected 0 bytes on stream wq2, got: %+v", si.State)
 } elseifsi.State.Bytes>maxBytes {
t.Fatalf("got more than our %d bytes limit on stream wq2: %+v", maxMsgs, si.State)
 }

returnnil
 })

select {
case<-doneChan2:
ss2.Drain()
case<-time.After(20*time.Second):
t.Fatalf("Did not receive completion signal")
 }
}

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

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

_, err:=js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"foo"},
Replicas: 3,
 })
require_NoError(t, err)

jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
Name: "my_consumer",
Replicas: 3,
 })

sub, err:=js.PullSubscribe("foo", "", nats.Bind("TEST", "my_consumer"))
require_NoError(t, err)

stepdown:=func() {
t.Helper()
_, err:=nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "TEST", "my_consumer"), nil, time.Second)
require_NoError(t, err)
c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
 }

publish:=func(wait time.Duration) {
t.Helper()
fori:=0; i<5; i++ {
_, err=js.Publish("foo", []byte("OK"))
require_NoError(t, err)
 }
msgs, err:=sub.Fetch(5, nats.MaxWait(wait))
require_NoError(t, err)
require_Equal(t, len(msgs), 5)
 }

// This should be fast as there's no deadline.
publish(time.Second)

// Now we're going to set the deadline.
deadline:=jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", time.Now().Add(time.Second*3))
c.waitOnAllCurrent()

// It will now take longer than 3 seconds.
publish(time.Second*5)
require_True(t, time.Now().After(deadline))

// The next set of publishes after the deadline should now be fast.
publish(time.Second)

// We'll kick the leader, but since we're after the deadline, this
// should still be fast.
stepdown()
publish(time.Second)

// Now we're going to do an update and then immediately kick the
// leader. The pause should still be in effect afterwards.
deadline=jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", time.Now().Add(time.Second*3))
c.waitOnAllCurrent()
publish(time.Second*5)
require_True(t, time.Now().After(deadline))

// The next set of publishes after the deadline should now be fast.
publish(time.Second)
}

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

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

_, err:=js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"push", "pull"},
Replicas: 3,
 })
require_NoError(t, err)

t.Run("PullConsumer", func(t*testing.T) {
_, err:=js.AddConsumer("TEST", &nats.ConsumerConfig{
Name: "pull_consumer",
 })
require_NoError(t, err)

sub, err:=js.PullSubscribe("pull", "", nats.Bind("TEST", "pull_consumer"))
require_NoError(t, err)

// This should succeed as there's no pause, so it definitely
// shouldn't take more than a second.
fori:=0; i<10; i++ {
_, err=js.Publish("pull", []byte("OK"))
require_NoError(t, err)
 }
msgs, err:=sub.Fetch(10, nats.MaxWait(time.Second))
require_NoError(t, err)
require_Equal(t, len(msgs), 10)

// Now we'll pause the consumer for 3 seconds.
deadline:=time.Now().Add(time.Second*3)
require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "pull_consumer", deadline).Equal(deadline))
c.waitOnAllCurrent()

// This should fail as we'll wait for only half of the deadline.
fori:=0; i<10; i++ {
_, err=js.Publish("pull", []byte("OK"))
require_NoError(t, err)
 }
_, err=sub.Fetch(10, nats.MaxWait(time.Until(deadline)/2))
require_Error(t, err, nats.ErrTimeout)

// This should succeed after a short wait, and when we're done,
// we should be after the deadline.
msgs, err=sub.Fetch(10)
require_NoError(t, err)
require_Equal(t, len(msgs), 10)
require_True(t, time.Now().After(deadline))

// This should succeed as there's no pause, so it definitely
// shouldn't take more than a second.
fori:=0; i<10; i++ {
_, err=js.Publish("pull", []byte("OK"))
require_NoError(t, err)
 }
msgs, err=sub.Fetch(10, nats.MaxWait(time.Second))
require_NoError(t, err)
require_Equal(t, len(msgs), 10)

require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "pull_consumer", time.Time{}).Equal(time.Time{}))
c.waitOnAllCurrent()

// This should succeed as there's no pause, so it definitely
// shouldn't take more than a second.
fori:=0; i<10; i++ {
_, err=js.Publish("pull", []byte("OK"))
require_NoError(t, err)
 }
msgs, err=sub.Fetch(10, nats.MaxWait(time.Second))
require_NoError(t, err)
require_Equal(t, len(msgs), 10)
 })

t.Run("PushConsumer", func(t*testing.T) {
ch:=make(chan*nats.Msg, 100)
_, err=js.ChanSubscribe("push", ch, nats.BindStream("TEST"), nats.ConsumerName("push_consumer"))
require_NoError(t, err)

// This should succeed as there's no pause, so it definitely
// shouldn't take more than a second.
fori:=0; i<10; i++ {
_, err=js.Publish("push", []byte("OK"))
require_NoError(t, err)
 }
fori:=0; i<10; i++ {
msg:=require_ChanRead(t, ch, time.Second)
require_NotEqual(t, msg, nil)
 }

// Now we'll pause the consumer for 3 seconds.
deadline:=time.Now().Add(time.Second*3)
require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "push_consumer", deadline).Equal(deadline))
c.waitOnAllCurrent()

// This should succeed after a short wait, and when we're done,
// we should be after the deadline.
fori:=0; i<10; i++ {
_, err=js.Publish("push", []byte("OK"))
require_NoError(t, err)
 }
fori:=0; i<10; i++ {
msg:=require_ChanRead(t, ch, time.Second*5)
require_NotEqual(t, msg, nil)
require_True(t, time.Now().After(deadline))
 }

// This should succeed as there's no pause, so it definitely
// shouldn't take more than a second.
fori:=0; i<10; i++ {
_, err=js.Publish("push", []byte("OK"))
require_NoError(t, err)
 }
fori:=0; i<10; i++ {
msg:=require_ChanRead(t, ch, time.Second)
require_NotEqual(t, msg, nil)
 }

require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "push_consumer", time.Time{}).Equal(time.Time{}))
c.waitOnAllCurrent()

// This should succeed as there's no pause, so it definitely
// shouldn't take more than a second.
fori:=0; i<10; i++ {
_, err=js.Publish("push", []byte("OK"))
require_NoError(t, err)
 }
fori:=0; i<10; i++ {
msg:=require_ChanRead(t, ch, time.Second)
require_NotEqual(t, msg, nil)
 }
 })
}

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

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

_, err:=js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"foo"},
Replicas: 3,
 })
require_NoError(t, err)

deadline:=time.Now().Add(time.Hour)
jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
Name: "my_consumer",
PauseUntil: &deadline,
Replicas: 3,
 })

fori:=0; i<10; i++ {
c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
c.waitOnAllCurrent()

for_, s:=rangec.servers {
stream, err:=s.gacc.lookupStream("TEST")
require_NoError(t, err)

consumer:=stream.lookupConsumer("my_consumer")
require_NotEqual(t, consumer, nil)

isLeader:=s.JetStreamIsConsumerLeader(globalAccountName, "TEST", "my_consumer")

consumer.mu.RLock()
hasTimer:=consumer.uptmr!=nil
consumer.mu.RUnlock()

require_Equal(t, isLeader, hasTimer)
 }

_, err=nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "TEST", "my_consumer"), nil, maxElectionTimeout)
require_NoError(t, err)
 }
}

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

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

_, err:=js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"TEST"},
Replicas: 3,
 })
require_NoError(t, err)

_, err=js.AddConsumer("TEST", &nats.ConsumerConfig{
Name: "CONSUMER",
Replicas: 3,
 })
require_NoError(t, err)

getConsumerInfo:=func() ConsumerInfo {
varciConsumerInfo
infoResp, err:=nc.Request("$JS.API.CONSUMER.INFO.TEST.CONSUMER", nil, time.Second)
require_NoError(t, err)
err=json.Unmarshal(infoResp.Data, &ci)
require_NoError(t, err)
returnci
 }

// Ensure we are not paused
require_False(t, getConsumerInfo().Paused)

// Use pause advisories to know when pause/resume is applied.
ch:=make(chan*nats.Msg, 10)
_, err=nc.ChanSubscribe(JSAdvisoryConsumerPausePre+".TEST.CONSUMER", ch)
require_NoError(t, err)

// Now we'll pause the consumer for 30 seconds.
deadline:=time.Now().Add(time.Second*30)
require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "CONSUMER", deadline).Equal(deadline))
require_ChanRead(t, ch, time.Second*2)
require_Len(t, len(ch), 0)

// Ensure the consumer reflects being paused
require_True(t, getConsumerInfo().Paused)

subj:=fmt.Sprintf("$JS.API.CONSUMER.PAUSE.%s.%s", "TEST", "CONSUMER")
_, err=nc.Request(subj, nil, time.Second)
require_NoError(t, err)
require_ChanRead(t, ch, time.Second*2)
require_Len(t, len(ch), 0)

// Ensure the consumer reflects being resumed
require_False(t, getConsumerInfo().Paused)
}

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

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

_, err:=js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"foo"},
Replicas: 3,
 })
require_NoError(t, err)

deadline:=time.Now().Add(time.Hour)
dsubj:="deliver_subj"

ci:=jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
Name: "my_consumer",
PauseUntil: &deadline,
Heartbeat: time.Millisecond*100,
DeliverSubject: dsubj,
 })
require_True(t, ci.Config.PauseUntil.Equal(deadline))

ch:=make(chan*nats.Msg, 10)
_, err=nc.ChanSubscribe(dsubj, ch)
require_NoError(t, err)

fori:=0; i<20; i++ {
msg:=require_ChanRead(t, ch, time.Millisecond*200)
require_Equal(t, msg.Header.Get("Status"), "100")
require_Equal(t, msg.Header.Get("Description"), "Idle Heartbeat")
 }
}

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

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

checkAdvisory:=func(msg*nats.Msg, shouldBePausedbool, deadline time.Time) {
t.Helper()
varadvisoryJSConsumerPauseAdvisory
require_NoError(t, json.Unmarshal(msg.Data, &advisory))
require_Equal(t, advisory.Stream, "TEST")
require_Equal(t, advisory.Consumer, "my_consumer")
require_Equal(t, advisory.Paused, shouldBePaused)
require_True(t, advisory.PauseUntil.Equal(deadline))
 }

_, err:=js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"foo"},
Replicas: 3,
 })
require_NoError(t, err)

ch:=make(chan*nats.Msg, 10)
_, err=nc.ChanSubscribe(JSAdvisoryConsumerPausePre+".TEST.my_consumer", ch)
require_NoError(t, err)

deadline:=time.Now().Add(time.Second)
jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
Name: "my_consumer",
PauseUntil: &deadline,
Replicas: 3,
 })

// First advisory should tell us that the consumer was paused
// on creation.
msg:=require_ChanRead(t, ch, time.Second*2)
checkAdvisory(msg, true, deadline)
require_Len(t, len(ch), 0) // Should only receive one advisory.

// The second one for the unpause.
msg=require_ChanRead(t, ch, time.Second*2)
checkAdvisory(msg, false, deadline)
require_Len(t, len(ch), 0) // Should only receive one advisory.

// Now we'll pause the consumer for a second using the API.
deadline=time.Now().Add(time.Second)
require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", deadline).Equal(deadline))

// Third advisory should tell us about the pause via the API.
msg=require_ChanRead(t, ch, time.Second*2)
checkAdvisory(msg, true, deadline)
require_Len(t, len(ch), 0) // Should only receive one advisory.

// Finally that should unpause.
msg=require_ChanRead(t, ch, time.Second*2)
checkAdvisory(msg, false, deadline)
require_Len(t, len(ch), 0) // Should only receive one advisory.

// Now we're going to set the deadline into the future so we can
// see what happens when we kick leaders or restart.
deadline=time.Now().Add(time.Hour)
require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", deadline).Equal(deadline))

// Setting the deadline should have generated an advisory.
msg=require_ChanRead(t, ch, time.Second)
checkAdvisory(msg, true, deadline)
require_Len(t, len(ch), 0) // Should only receive one advisory.

// Try to kick the consumer leader.
srv:=c.consumerLeader(globalAccountName, "TEST", "my_consumer")
srv.JetStreamStepdownConsumer(globalAccountName, "TEST", "my_consumer")
c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")

// This shouldn't have generated an advisory.
require_NoChanRead(t, ch, time.Second)
}

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

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

checkTimer:=func(s*Server) {
stream, err:=s.gacc.lookupStream("TEST")
require_NoError(t, err)

consumer:=stream.lookupConsumer("my_consumer")
require_NotEqual(t, consumer, nil)

consumer.mu.RLock()
timer:=consumer.uptmr
consumer.mu.RUnlock()
require_True(t, timer!=nil)
 }

_, err:=js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"foo"},
Replicas: 3,
 })
require_NoError(t, err)

deadline:=time.Now().Add(time.Hour)
jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
Name: "my_consumer",
PauseUntil: &deadline,
Replicas: 3,
 })

// First try with just restarting the consumer leader.
srv:=c.consumerLeader(globalAccountName, "TEST", "my_consumer")
srv.Shutdown()
c.restartServer(srv)
c.waitOnAllCurrent()
c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
leader:=c.consumerLeader(globalAccountName, "TEST", "my_consumer")
require_True(t, leader!=nil)
checkTimer(leader)

// Then try restarting the entire cluster.
c.stopAll()
c.restartAllSamePorts()
c.waitOnAllCurrent()
c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
leader=c.consumerLeader(globalAccountName, "TEST", "my_consumer")
require_True(t, leader!=nil)
checkTimer(leader)
}

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

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

_, err:=js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"foo"},
Storage: nats.MemoryStorage,
Retention: nats.WorkQueuePolicy,
Replicas: 3,
 })
require_NoError(t, err)

// First call is just to create the pull subscribers.
_, err=js.PullSubscribe("foo", "dlc")
require_NoError(t, err)

require_NoError(t, js.DeleteConsumer("TEST", "dlc"))

// Now delete the stream.
require_NoError(t, js.DeleteStream("TEST"))

// Now make sure we cleaned up the NRG directories for the stream and consumer.
checkFor(t, 2*time.Second, 500*time.Millisecond, func() error {
varnumConsumers, numStreamsint
for_, s:=rangec.servers {
sd:=s.JetStreamConfig().StoreDir
nd:=filepath.Join(sd, "$SYS", "_js_")
f, err:=os.Open(nd)
iferr!=nil {
returnerr
 }
dirs, err:=f.ReadDir(-1)
iferr!=nil {
returnerr
 }
for_, fi:=rangedirs {
ifstrings.HasPrefix(fi.Name(), "C-") {
numConsumers++
 } elseifstrings.HasPrefix(fi.Name(), "S-") {
numStreams++
 }
 }
f.Close()
 }

ifnumConsumers!=0 {
returnfmt.Errorf("expected 0 consumers, got %d", numConsumers)
 }
ifnumStreams!=0 {
returnfmt.Errorf("expected 0 streams, got %d", numStreams)
 }
returnnil
 })
}

// https://github.com/nats-io/nats-server/issues/4878
funcTestClusteredInterestConsumerFilterEdit(t*testing.T) {
c:=createJetStreamClusterExplicit(t, "R3S", 3)
deferc.shutdown()

s:=c.randomNonLeader()

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

_, err:=js.AddStream(&nats.StreamConfig{
Name: "INTEREST",
Retention: nats.InterestPolicy,
Subjects: []string{"interest.>"},
Replicas: 3,
 })
require_NoError(t, err)

_, err=js.AddConsumer("INTEREST", &nats.ConsumerConfig{
Durable: "C0",
FilterSubject: "interest.>",
AckPolicy: nats.AckExplicitPolicy,
 })
require_NoError(t, err)

fori:=0; i<10; i++ {
_, err=js.Publish(fmt.Sprintf("interest.%d", i), []byte(strconv.Itoa(i)))
require_NoError(t, err)
 }

// we check we got 10 messages
nfo, err:=js.StreamInfo("INTEREST")
require_NoError(t, err)
ifnfo.State.Msgs!=10 {
t.Fatalf("expected 10 messages got %d", nfo.State.Msgs)
 }

// now we lower the consumer interest from all subjects to 1,
// then check the stream state and check if interest behavior still works
_, err=js.UpdateConsumer("INTEREST", &nats.ConsumerConfig{
Durable: "C0",
FilterSubject: "interest.1",
AckPolicy: nats.AckExplicitPolicy,
 })
require_NoError(t, err)

// we should now have only one message left
nfo, err=js.StreamInfo("INTEREST")
require_NoError(t, err)
ifnfo.State.Msgs!=1 {
t.Fatalf("expected 1 message got %d", nfo.State.Msgs)
 }
}

funcTestJetStreamClusterDoubleAckRedelivery(t*testing.T) {
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, "R3F", 3)
deferc.shutdown()
for_, s:=rangec.servers {
s.optsMu.Lock()
s.opts.LameDuckDuration=15*time.Second
s.opts.LameDuckGracePeriod=-15*time.Second
s.optsMu.Unlock()
 }
s:=c.randomNonLeader()

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

sc, err:=js.AddStream(&nats.StreamConfig{
Name: "LIMITS",
Subjects: []string{"foo.>"},
Replicas: 3,
Storage: nats.FileStorage,
 })
require_NoError(t, err)

stepDown:=func() {
_, err=nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, sc.Config.Name), nil, time.Second)
 }

ctx, cancel:=context.WithCancel(context.Background())
defercancel()

varwg sync.WaitGroup