Merge pull request #2832 from nats-io/nak-nowait

A true no wait pull request was not considering redeliveries.

server/consumer.go

@@ -2374,7 +2374,7 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 
 	// If the request is for noWait and we have pending requests already, check if we have room.
 	if noWait {
-		msgsPending := o.adjustedPending()
+		msgsPending := o.adjustedPending() + uint64(len(o.rdq))
 		// If no pending at all, decide what to do with request.
 		// If no expires was set then fail.
 		if msgsPending == 0 && expires.IsZero() {

server/filestore.go

@@ -4900,9 +4900,8 @@ func (o *consumerFileStore) UpdateDelivered(dseq, sseq, dc uint64, ts int64) err
 			if p = o.state.Pending[sseq]; p != nil {
 				p.Sequence, p.Timestamp = dseq, ts
 			}
-		}
-		// Add to pending if needed.
-		if p == nil {
+		} else {
+			// Add to pending.
 			o.state.Pending[sseq] = &Pending{dseq, ts}
 		}
 		// Update delivered as needed.

server/jetstream_test.go

@@ -14545,6 +14545,75 @@ func TestJetStreamConsumerUpdateSurvival(t *testing.T) {
 	}
 }
 
+func TestJetStreamNakRedeliveryWithNoWait(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	_, err = js.Publish("foo", []byte("NAK"))
+	require_NoError(t, err)
+
+	ccReq := &CreateConsumerRequest{
+		Stream: "TEST",
+		Config: ConsumerConfig{
+			Durable:    "dlc",
+			AckPolicy:  AckExplicit,
+			MaxDeliver: 3,
+			AckWait:    time.Minute,
+			BackOff:    []time.Duration{5 * time.Second, 10 * time.Second},
+		},
+	}
+	// Do by hand for now until Go client catches up.
+	req, err := json.Marshal(ccReq)
+	require_NoError(t, err)
+	resp, err := nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", "dlc"), req, time.Second)
+	require_NoError(t, err)
+	var ccResp JSApiConsumerCreateResponse
+	err = json.Unmarshal(resp.Data, &ccResp)
+	require_NoError(t, err)
+	if ccResp.Error != nil {
+		t.Fatalf("Unexpected error: %+v", ccResp.Error)
+	}
+
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "TEST", "dlc")
+	m, err := nc.Request(rsubj, nil, time.Second)
+	require_NoError(t, err)
+
+	// NAK this message.
+	delay, err := json.Marshal(&ConsumerNakOptions{Delay: 500 * time.Millisecond})
+	require_NoError(t, err)
+	dnak := []byte(fmt.Sprintf("%s  %s", AckNak, delay))
+	m.Respond(dnak)
+
+	// This message should come back to us after 500ms. If we do a one-shot request, with NoWait and Expires
+	// this will do the right thing and we get the message.
+	// What we want to test here is a true NoWait request with Expires==0 and eventually seeing the message be redelivered.
+	expires := time.Now().Add(time.Second)
+	for time.Now().Before(expires) {
+		m, err = nc.Request(rsubj, []byte(`{"batch":1, "no_wait": true}`), time.Second)
+		require_NoError(t, err)
+		if len(m.Data) > 0 {
+			// We got our message, so we are good.
+			return
+		}
+		// So we do not spin.
+		time.Sleep(100 * time.Millisecond)
+	}
+	t.Fatalf("Did not get the message in time")
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Simple JetStream Benchmarks
 ///////////////////////////////////////////////////////////////////////////

server/norace_test.go

@@ -3332,7 +3332,8 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 		}
 	}
 	// Make sure acks processed.
-	nc.Flush()
+	time.Sleep(200 * time.Millisecond)
+	nc.Close()
 
 	// Check consumer consistency.
 	checkConsumerWith := func(delivered, ackFloor uint64, ackPending int) {
@@ -3383,6 +3384,7 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 	node := o.raftNode().(*raft)
 	fs := node.wal.(*fileStore)
 	fcfg, cfg := fs.fcfg, fs.cfg.StreamConfig
+	// Stop all the servers.
 	c.stopAll()
 
 	// Manipulate directly with cluster down.
@@ -3451,8 +3453,13 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 	state = fs.State()
 	fs.Truncate(state.FirstSeq)
 
+	sub, err = js.PullSubscribe("foo", "dlc")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
 	// This will cause us to stepdown and truncate our WAL.
-	fetchMsgs(t, sub, 100, 50*time.Millisecond)
+	fetchMsgs(t, sub, 100, 5*time.Second)
 
 	checkFor(t, 20*time.Second, 500*time.Millisecond, func() error {
 		// Make sure we changed leaders.