Merge pull request #3658 from nats-io/fix-3657

[FIXED] Server might crash on pull consumer delete.

server/consumer.go

@@ -3130,10 +3130,14 @@ func (o *consumer) suppressDeletion() {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
+	if o.closed {
+		return
+	}
+
 	if o.isPushMode() && o.dtmr != nil {
 		// if dtmr is not nil we have started the countdown, simply reset to threshold.
 		o.dtmr.Reset(o.dthresh)
-	} else if o.isPullMode() {
+	} else if o.isPullMode() && o.waiting != nil {
 		// Pull mode always has timer running, just update last on waiting queue.
 		o.waiting.last = time.Now()
 	}

server/jetstream_test.go

@@ -19827,3 +19827,37 @@ func TestJetStreamMsgIDHeaderCollision(t *testing.T) {
 
 	require_True(t, si.State.Msgs == 5)
 }
+
+// https://github.com/nats-io/nats-server/issues/3657
+func TestJetStreamServerCrashOnPullConsumerDeleteWithInactiveThresholdAfterAck(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	sendStreamMsg(t, nc, "foo", "msg")
+
+	sub, err := js.PullSubscribe("foo", "dlc", nats.InactiveThreshold(10*time.Second))
+	require_NoError(t, err)
+
+	msgs := fetchMsgs(t, sub, 1, time.Second)
+	require_True(t, len(msgs) == 1)
+	msgs[0].Ack()
+	err = js.DeleteConsumer("TEST", "dlc")
+	require_NoError(t, err)
+
+	// If server crashes this will fail.
+	_, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+}