(2.14) [FIXED] Consumer send 404 No Messages on EOS after delivering messages

Signed-off-by: Maurice van Veen <[email protected]>

Author: MauriceVanVeen

server/consumer.go

@@ -4536,12 +4536,12 @@ func (o *consumer) processWaiting(eos bool) (int, int, int, time.Time) {
 				}
 			}
 			// Normally it's a timeout.
-			if expires || !wr.noWait || wr.d > 0 {
+			if expires {
 				hdr := fmt.Appendf(nil, "NATS/1.0 408 Request Timeout\r\n%s: %d\r\n%s: %d\r\n\r\n", JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b)
 				o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 				wr = remove(pre, wr)
 				continue
-			} else if wr.expires.IsZero() {
+			} else if wr.expires.IsZero() || wr.d > 0 {
 				// But if we're NoWait without expiry, we've reached the end of the stream, and we've not delivered any messages.
 				// Return no messages instead, which is the same as if we'd rejected the pull request initially.
 				hdr := fmt.Appendf(nil, "NATS/1.0 404 No Messages\r\n\r\n")

server/jetstream_consumer_test.go

@@ -10381,3 +10381,46 @@ func TestJetStreamConsumerNoWaitNoMessagesOnEos(t *testing.T) {
 	require_Equal(t, msg.Header.Get("Status"), "404")
 	require_Equal(t, msg.Header.Get("Description"), "No Messages")
 }
+
+// https://github.com/nats-io/nats-server/issues/5373
+func TestJetStreamConsumerNoWaitNoMessagesOnEosWithDeliveredMsgs(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"foo"}})
+	require_NoError(t, err)
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "CONSUMER"})
+	require_NoError(t, err)
+
+	_, err = js.Publish("foo", []byte("msg"))
+	require_NoError(t, err)
+
+	sub, err := nc.SubscribeSync("reply")
+	require_NoError(t, err)
+	defer sub.Drain()
+	require_NoError(t, nc.Flush())
+
+	mset, err := s.globalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	o := mset.lookupConsumer("CONSUMER")
+	require_NotNil(t, o)
+
+	req := &JSApiConsumerGetNextRequest{NoWait: true, Batch: 2}
+	jreq, err := json.Marshal(req)
+	require_NoError(t, err)
+	o.processNextMsgRequest("reply", jreq)
+
+	msg, err := sub.NextMsg(time.Second)
+	require_NoError(t, err)
+	require_Equal(t, msg.Subject, "foo")
+	require_Equal(t, string(msg.Data), "msg")
+
+	// We requested two messages but the stream only contained 1.
+	msg, err = sub.NextMsg(time.Second)
+	require_NoError(t, err)
+	require_Equal(t, msg.Header.Get("Status"), "404")
+	require_Equal(t, msg.Header.Get("Description"), "No Messages")
+}