Repository: nats-io/nats-server

Issue:
FetchNoWait with batch size does not work with v2.11.4

### Suspected Bug in NATS v2.11.4 (Cluster Mode) When Using `FetchNoWait` with Large Batch Size

We use FetchNoWait with batch size 1000 to fetch messages in the stream. This setup worked fine with earlier NATS versions with single-node/multiple-nodes cluster. 

However, after upgrading to NATS v2.11.4 and deploying a 3-node cluster, we observed that FetchNoWait does not work if consumer replica > 1 and the batch size is bigger than the total messages of that subject.

It only works if either
- Consumer replica is 1, or
- The maximum batch size is the total messages of that subject in the stream.

#### Observations

## 1 node

* Stream information

```
Information for Stream EVENTS created 2025-06-09 16:47:36

              Subjects: events.>
              Replicas: 1
               Storage: File

Options:

             Retention: Limits
       Acknowledgments: true
        Discard Policy: Old
      Duplicate Window: 2m0s
     Allows Msg Delete: true
          Allows Purge: true
        Allows Rollups: false

Limits:

      Maximum Messages: unlimited
   Maximum Per Subject: unlimited
         Maximum Bytes: unlimited
           Maximum Age: unlimited
  Maximum Message Size: unlimited
     Maximum Consumers: unlimited

Metadata:

           _nats.level: 1
       _nats.req.level: 0
             _nats.ver: 2.11.4

State:

              Messages: 105
                 Bytes: 5.1 KiB
        First Sequence: 1 @ 2025-06-09 16:47:58
         Last Sequence: 105 @ 2025-06-09 16:48:08
      Active Consumers: 0
    Number of Subjects: 1
```

* The stream has 105 messages from the subject **events.new**

![Image](https://github.com/user-attachments/assets/e57e13a0-e12f-4e32-aecf-752733754ae2)

* `FetchNoWait` with batch size 1000

```
Information for Consumer EVENTS > consumer1 created 2025-06-09T17:09:34+12:00

Configuration:

                    Name: consumer1
               Pull Mode: true
          Filter Subject: events.new
          Deliver Policy: From Sequence 50
              Ack Policy: All
                Ack Wait: 30.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512
      Inactive Threshold: 15m0s
                Replicas: 1

Metadata:

             _nats.level: 1
         _nats.req.level: 0
               _nats.ver: 2.11.4

State:

  Last Delivered Message: Consumer sequence: 56 Stream sequence: 105 Last delivery: 44.38s ago
    Acknowledgment Floor: Consumer sequence: 56 Stream sequence: 105 Last Ack: 44.38s ago
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 0 of maximum 512
```

* We can see that all messages has been delivered to this consumer

## 3 nodes

* Stream has 170 messages on subject **events.new**

```
Information for Stream EVENTS created 2025-06-09 17:12:49

              Subjects: events.>
              Replicas: 3
               Storage: File

Options:

             Retention: Limits
       Acknowledgments: true
        Discard Policy: Old
      Duplicate Window: 2m0s
     Allows Msg Delete: true
          Allows Purge: true
        Allows Rollups: false

Limits:

      Maximum Messages: unlimited
   Maximum Per Subject: unlimited
         Maximum Bytes: unlimited
           Maximum Age: unlimited
  Maximum Message Size: unlimited
     Maximum Consumers: unlimited

Metadata:

           _nats.level: 1
       _nats.req.level: 0
             _nats.ver: 2.11.4

Cluster Information:

                  Name: nats-cluster
                Leader: nats-2
               Replica: nats-1, current, seen 119ms ago
               Replica: nats-3, current, seen 119ms ago

State:

              Messages: 170
                 Bytes: 8.4 KiB
        First Sequence: 1 @ 2025-06-09 17:12:49
         Last Sequence: 170 @ 2025-06-09 17:13:06
      Active Consumers: 0
    Number of Subjects: 1
```

![Image](https://github.com/user-attachments/assets/09e7ff56-c197-421a-894e-5d63a79de636)

Running `FetchNoWait` with batch size is 1000, we can see that the stream delivered 170 messages but consumer has not acked any messages as no messages returned

```
Information for Consumer EVENTS > consumerrr created 2025-06-09T17:15:32+12:00

Configuration:

                    Name: consumerrr
               Pull Mode: true
          Filter Subject: events.new
          Deliver Policy: From Sequence 1
              Ack Policy: All
                Ack Wait: 30.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512
      Inactive Threshold: 15m0s
                Replicas: 3

Metadata:

             _nats.level: 1
         _nats.req.level: 0
               _nats.ver: 2.11.4

Cluster Information:

                    Name: nats-cluster
                  Leader: nats-2
                 Replica: nats-1, current, seen 745ms ago
                 Replica: nats-3, current, seen 745ms ago

State:

  Last Delivered Message: Consumer sequence: 170 Stream sequence: 170 Last delivery: 22.74s ago
    Acknowledgment Floor: Consumer sequence: 0 Stream sequence: 0
        Outstanding Acks: 170 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 0 of maximum 512
```

If I set the batchsize equal to the current total messages in the subject, then it can fetch

```
Information for Consumer EVENTS > consumerrr created 2025-06-09T17:15:32+12:00

Configuration:

                    Name: consumerrr
               Pull Mode: true
          Filter Subject: events.new
          Deliver Policy: From Sequence 1
              Ack Policy: All
                Ack Wait: 30.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512
      Inactive Threshold: 15m0s
                Replicas: 3

Metadata:

               _nats.ver: 2.11.4
             _nats.level: 1
         _nats.req.level: 0

Cluster Information:

                    Name: nats-cluster
                  Leader: nats-2
                 Replica: nats-1, current, seen 396ms ago
                 Replica: nats-3, current, seen 396ms ago

State:

  Last Delivered Message: Consumer sequence: 340 Stream sequence: 170 Last delivery: 5.94s ago
    Acknowledgment Floor: Consumer sequence: 340 Stream sequence: 170 Last Ack: 5.93s ago
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 0 of maximum 512
```

### Expected behavior

We expect that this FetchNoWait with batch size should works if batch size > total messages in the subjects?

### Server and client version

client version v1.42.0 and server version 2.11.4

### Host environment

_No response_

### Steps to reproduce

_No response_

Discussion:
@souravagrawal: can you please share your consumer client code once ? @Jarema: With quick test, I was able to pull messages following your scenario, so we would need a code sample to  better understand what is happening.@Helenpham1097: Hi, this is my code example.

With a cluster of 3 nodes, if I create a subscription with 1 replica, then it can fetch as well.


```

// create a test stream with 3 replicas
func create() {
	// Connect to NATS
	nc, err := connect()

	if err != nil {
		log.Fatal(err)
	}
	defer nc.Close()

	// Enable JetStream
	js, err := nc.JetStream()
	if err != nil {
		log.Fatal(err)
	}

	// Add a stream if it doesn't exist
	streamConfig := &nats.StreamConfig{
		Name:     "EVENTS",
		Subjects: []string{"events.>"},
		Storage:  nats.FileStorage,
		Replicas: 3,
	}

	_, err = js.AddStream(streamConfig)
	if err != nil && !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
		log.Fatal("Failed to create stream:", err)
	} else {
		log.Println("Stream created (or already exists)", nc.ConnectedAddr())
	}
}

func connect() (*nats.Conn, error) {
	conn, err := nats.Connect("nats://nats-cluster.local:4222",
		nats.MaxReconnects(10),
		nats.ReconnectWait(time.Second*2),
		nats.RetryOnFailedConnect(true),
		nats.Timeout(time.Millisecond*500),
		nats.DisconnectHandler(func(nc *nats.Conn) {
			log.Printf("Disconnected from NATS: %v", nc.LastError())
		}),
		nats.ReconnectHandler(func(nc *nats.Conn) { // triiger if task stops.
			log.Printf("Reconnected to NATS at %s: %s", nc.ConnectedUrl(), nc.ConnectedServerName())
		}),
		nats.ClosedHandler(func(nc *nats.Conn) {
			log.Printf("Connection closed: %v", nc.LastError())
		}),
		nats.ErrorHandler(func(conn *nats.Conn, sub *nats.Subscription, err error) {
			log.Printf("nats: received error on connection to %s: %v", conn.ConnectedServerName(), err)
		}),
	)

	if err != nil {
		return nil, err
	}

	return conn, nil
}

func resubscribe(ctx context.Context, js jetstream.JetStream, sub jetstream.Consumer) (jetstream.Consumer, error) {
	sub, err := subscribe(ctx, js)
	if err != nil {
		return nil, fmt.Errorf("creating new consumer: %w", err)
	}

	return sub, nil
}

func subscribe(ctx context.Context, js jetstream.JetStream) (jetstream.Consumer, error) {
	opts := jetstream.ConsumerConfig{
		Name:              "consumer1",
		InactiveThreshold: 15 * time.Minute,
		FilterSubject:     "events.new",
		AckPolicy:         jetstream.AckAllPolicy,
		Replicas:          3,
		OptStartSeq:       1, // Start at the sequence we got from the stream info
		DeliverPolicy:     jetstream.DeliverByStartSequencePolicy,
	}

	var sub jetstream.Consumer
	for i := 1; i <= 10; i++ {
		subscribeCtx, cancel := context.WithTimeout(ctx, time.Second*2)
		var err error
		if sub, err = js.CreateOrUpdateConsumer(subscribeCtx, "EVENTS", opts); err != nil {
			fmt.Printf("invalidator: failed to create or update consumer %s: %v \n", opts.Name, err)
			if i == 10 {
				cancel()
				return sub, fmt.Errorf("creating or updating consumer %s: %w", opts.Name, err)
			}

			cancel()
			time.Sleep(500 * time.Millisecond)
			continue
		}

		cancel()
		break
	}

	return sub, nil
}

func consumer() {
	nc, err := connect()
	if err != nil {
		log.Fatal(err)
	}
	defer nc.Close()

	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatalf("Error create stream: %v", err)
	}

	log.Println("Subscriber stream ", nc.ConnectedAddr(), nc.ConnectedServerName())

	var sub jetstream.Consumer

	sub, err = resubscribe(context.Background(), nc, js, sub)
	if err != nil {
		log.Fatal(err)
	}

        // Could not fetch any messages if I set the batch size larger than to total message in the subject. Let say 1000
	batch, err := sub.FetchNoWait(170)
	if err != nil {
		log.Println("Error while fetching messages", err)
	}

	if batch == nil {
		log.Println("No messages")
		return
	}

	count := 0
	for msg := range batch.Messages() {
		meta, err := msg.Metadata()
		if err != nil {
			log.Println("Error while parsing metadata", err)
			continue
		}

		log.Println("Got message", string(msg.Data()), "sequence ", meta.Sequence)

		if err := msg.Ack(); err != nil {
			fmt.Printf("Failed to ack message: %v", err)
			continue
		}
		count++
	}

	log.Println("Total messages acked", count)
}
```

Also this is a docker image I use

```
FROM nats:2.11.4-alpine AS nats-upstream

FROM golang:1.24.3-alpine AS go-build
RUN go install github.com/nats-io/natscli/nats@v0.2.3 && \
    go install github.com/nats-io/nats-top@v0.6.3

FROM golang:1.24.3-alpine

COPY --from=nats-upstream /usr/local/bin/nats-server /usr/local/bin/nats-server
COPY --from=go-build /go/bin/* /go/bin/

# Set the GOPATH in the environment
ENV GOPATH=/go/ CGO_ENABLED=0 GOOS=linux GOARCH=arm64

RUN apk update && apk add --no-cache curl git ca-certificates && update-ca-certificates && \
	echo -e "nameserver 8.8.8.8\nameserver 8.8.4.4\n" > /go/bin/resolv.conf && \
	mkdir -p $GOPATH/example
# Set our workdir (aka the workspace or GOPATH)
WORKDIR $GOPATH/example
````

Not sure if any wrong from my image but my old image works with version `nats:2.10.7`, golang `1.21.3` , `nats-cli 0.0.35` and `nats-top 0.6.0`)

Thanks for your helps.
@Helenpham1097: We use go client library version `v1.42.0`@derekcollison: Looping in @piotrpio who might be able to take a look.@piotrpio: I was able to reproduce, I'll post an update here when I'll have more info.@MauriceVanVeen: Will be fully fixed once https://github.com/nats-io/nats-server/pull/7046 is released. Not just for `NoWait`, but also for `Expires`.

Solution patch:
diff --git a/server/consumer.go b/server/consumer.go
index d101d54c..ddc0f6b8 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -4479,6 +4479,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			delay    time.Duration
 			sz       int
 			wrn, wrb int
+			wrNoWait bool
 		)
 
 		o.mu.Lock()
@@ -4557,7 +4558,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		if o.isPushMode() {
 			dsubj = o.dsubj
 		} else if wr := o.nextWaiting(sz); wr != nil {
-			wrn, wrb = wr.n, wr.b
+			wrn, wrb, wrNoWait = wr.n, wr.b, wr.noWait
 			dsubj = wr.reply
 			if o.cfg.PriorityPolicy == PriorityPinnedClient {
 				// FIXME(jrm): Can we make this prettier?
@@ -4632,7 +4633,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		}
 
 		// Do actual delivery.
-		o.deliverMsg(dsubj, ackReply, pmsg, dc, rp)
+		o.deliverMsg(dsubj, ackReply, pmsg, dc, rp, wrNoWait)
 
 		// If given request fulfilled batch size, but there are still pending bytes, send information about it.
 		if wrn <= 0 && wrb > 0 {
@@ -4831,7 +4832,7 @@ func convertToHeadersOnly(pmsg *jsPubMsg) {
 
 // Deliver a msg to the consumer.
 // Lock should be held and o.mset validated to be non-nil.
-func (o *consumer) deliverMsg(dsubj, ackReply string, pmsg *jsPubMsg, dc uint64, rp RetentionPolicy) {
+func (o *consumer) deliverMsg(dsubj, ackReply string, pmsg *jsPubMsg, dc uint64, rp RetentionPolicy, wrNoWait bool) {
 	if o.mset == nil {
 		pmsg.returnToPool()
 		return
@@ -4867,7 +4868,9 @@ func (o *consumer) deliverMsg(dsubj, ackReply string, pmsg *jsPubMsg, dc uint64,
 	// If we're replicated we MUST only send the message AFTER we've got quorum for updating
 	// delivered state. Otherwise, we could be in an invalid state after a leader change.
 	// We can send immediately if not replicated, not using acks, or using flow control (incompatible).
-	if o.node == nil || ap == AckNone || o.cfg.FlowControl {
+	// TODO(mvv): If NoWait we also bypass replicating first.
+	//  Ideally we'd only send the NoWait request timeout after replication and delivery.
+	if o.node == nil || ap == AckNone || o.cfg.FlowControl || wrNoWait {
 		o.outq.send(pmsg)
 	} else {
 		o.addReplicatedQueuedMsg(pmsg)



Test patch:
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index fbdd21ec..bbb2ba0c 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -9642,3 +9642,46 @@ func TestJetStreamConsumerStateAlwaysFromStore(t *testing.T) {
 	require_Equal(t, ci.Delivered.Stream, 1)
 	require_Equal(t, ci.AckFloor.Stream, 1)
 }
+
+func TestJetStreamConsumerPullNoWaitBatchLargerThanPending(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:       "C",
+		AckPolicy:     nats.AckExplicitPolicy,
+		FilterSubject: "foo",
+	})
+	require_NoError(t, err)
+
+	req := JSApiConsumerGetNextRequest{Batch: 10, NoWait: true}
+
+	for range 5 {
+		_, err := js.Publish("foo", []byte("OK"))
+		require_NoError(t, err)
+	}
+
+	sub := sendRequest(t, nc, "rply", req)
+	defer sub.Unsubscribe()
+
+	// Should get all 5 messages.
+	// TODO(mvv): Currently bypassing replicating first, need to figure out
+	//  how to send NoWait's request timeout after replication.
+	for range 5 {
+		msg, err := sub.NextMsg(time.Second)
+		require_NoError(t, err)
+		if len(msg.Data) == 0 && msg.Header != nil {
+			t.Fatalf("Expected data, got: %s", msg.Header.Get("Description"))
+		}
+	}
+}