Repository: nats-io/nats-server
Pull request #4697 (2023-10-24 13:17:33)

Extend the term advisory with a reason
 * Support getting a reason from the client doing a term
 * Supply some reasons when terminating messages during limits processing

Signed-off-by: R.I.Pienaar <rip@devco.net>



ripienaar: > I like the concept of a reason, but not sure of approach of simply tacking a random string onto the term msg itself.

Took my queue from AckNext which takes a simple form and a JSON form for more complex stuff, if you have a prefered alternative approach happy to change here
ripienaar: Updated and rebased

Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 4beadeef..6c5d5091 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -272,6 +272,10 @@ var (
 	AckNext = []byte("+NXT")
 	// Terminate delivery of the message.
 	AckTerm = []byte("+TERM")
+
+	// reasons to supply when terminating messages using limits
+	ackTermLimitsReason        = "Message deleted by stream limits"
+	ackTermUnackedLimitsReason = "Unacknowledged message was deleted"
 )
 
 // Calculate accurate replicas for the consumer config with the parent stream config.
@@ -1991,8 +1995,12 @@ func (o *consumer) processAck(subject, reply string, hdr int, rmsg []byte) {
 		o.processNak(sseq, dseq, dc, msg)
 	case bytes.Equal(msg, AckProgress):
 		o.progressUpdate(sseq)
-	case bytes.Equal(msg, AckTerm):
-		o.processTerm(sseq, dseq, dc)
+	case bytes.HasPrefix(msg, AckTerm):
+		var reason string
+		if buf := msg[len(AckTerm):]; len(buf) > 0 {
+			reason = string(bytes.TrimSpace(buf))
+		}
+		o.processTerm(sseq, dseq, dc, reason)
 	}
 
 	// Ack the ack if requested.
@@ -2319,7 +2327,7 @@ func (o *consumer) processNak(sseq, dseq, dc uint64, nak []byte) {
 }
 
 // Process a TERM
-func (o *consumer) processTerm(sseq, dseq, dc uint64) {
+func (o *consumer) processTerm(sseq, dseq, dc uint64, reason string) {
 	// Treat like an ack to suppress redelivery.
 	o.processAckMsg(sseq, dseq, dc, false)
 
@@ -2338,6 +2346,7 @@ func (o *consumer) processTerm(sseq, dseq, dc uint64) {
 		ConsumerSeq: dseq,
 		StreamSeq:   sseq,
 		Deliveries:  dc,
+		Reason:      reason,
 		Domain:      o.srv.getOpts().JetStreamDomain,
 	}
 
@@ -3650,7 +3659,7 @@ func (o *consumer) checkAckFloor() {
 			o.mu.RUnlock()
 			// If it was pending for us, get rid of it.
 			if isPending {
-				o.processTerm(seq, p.Sequence, rdc)
+				o.processTerm(seq, p.Sequence, rdc, ackTermLimitsReason)
 			}
 		}
 	} else if numPending > 0 {
@@ -3675,7 +3684,7 @@ func (o *consumer) checkAckFloor() {
 
 		for i := 0; i < len(toTerm); i += 3 {
 			seq, dseq, rdc := toTerm[i], toTerm[i+1], toTerm[i+2]
-			o.processTerm(seq, dseq, rdc)
+			o.processTerm(seq, dseq, rdc, ackTermLimitsReason)
 		}
 	}
 
@@ -5151,11 +5160,10 @@ func (o *consumer) decStreamPending(sseq uint64, subj string) {
 	o.mu.Unlock()
 
 	// If it was pending process it like an ack.
-	// TODO(dlc) - we could do a term here instead with a reason to generate the advisory.
 	if wasPending {
 		// We could have lock for stream so do this in a go routine.
 		// TODO(dlc) - We should do this with ipq vs naked go routines.
-		go o.processTerm(sseq, p.Sequence, rdc)
+		go o.processTerm(sseq, p.Sequence, rdc, ackTermUnackedLimitsReason)
 	}
 }
 
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 2e1c7313..ab7c6191 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -134,6 +134,7 @@ type JSConsumerDeliveryTerminatedAdvisory struct {
 	ConsumerSeq uint64 `json:"consumer_seq"`
 	StreamSeq   uint64 `json:"stream_seq"`
 	Deliveries  uint64 `json:"deliveries"`
+	Reason      string `json:"reason,omitempty"`
 	Domain      string `json:"domain,omitempty"`
 }
 
diff --git a/server/mqtt.go b/server/mqtt.go
index b75a4a7c..41f3eee7 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -598,9 +598,9 @@ func validateMQTTOptions(o *Options) error {
 	if mo.Port == 0 {
 		return nil
 	}
-	// We have to force the server name to be explicitly set and be unique when
-	// in cluster mode.
-	if o.ServerName == _EMPTY_ && (o.Cluster.Port != 0 || o.Gateway.Port != 0) {
+	// We have to force the server name to be explicitly set. There are conditions
+	// where we need a unique, repeatable name.
+	if o.ServerName == _EMPTY_ {
 		return errMQTTServerNameMustBeSet
 	}
 	// If there is a NoAuthUser, we need to have Users defined and


Test patch:
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index faeb9cee..8cca50e1 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -2935,7 +2935,7 @@ func TestJetStreamWorkQueueTerminateDelivery(t *testing.T) {
 			// We should get 1 back.
 			m := getMsg(1, 2)
 			// Now terminate
-			m.Respond(AckTerm)
+			m.Respond([]byte(fmt.Sprintf("%s with reason", string(AckTerm))))
 			time.Sleep(ackWait * 2)
 
 			// We should get 2 here, not 1 since we have indicated we wanted to terminate.
@@ -2963,6 +2963,9 @@ func TestJetStreamWorkQueueTerminateDelivery(t *testing.T) {
 			if adv.Deliveries != 2 {
 				t.Fatalf("Expected delivery count of %d, got %d", 2, adv.Deliveries)
 			}
+			if adv.Reason != "with reason" {
+				t.Fatalf("Advisory did not have a reason")
+			}
 		})
 	}
 }
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index e92264ea..d521dce6 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -375,9 +375,6 @@ func testMQTTDefaultTLSOptions(t *testing.T, verify bool) *Options {
 
 func TestMQTTServerNameRequired(t *testing.T) {
 	conf := createConfFile(t, []byte(`
-		cluster {
-			port: -1
-		}
 		mqtt {
 			port: -1
 		}
@@ -389,19 +386,6 @@ func TestMQTTServerNameRequired(t *testing.T) {
 	if _, err := NewServer(o); err == nil || err.Error() != errMQTTServerNameMustBeSet.Error() {
 		t.Fatalf("Expected error about requiring server name to be set, got %v", err)
 	}
-
-	conf = createConfFile(t, []byte(`
-		mqtt {
-			port: -1
-		}
-	`))
-	o, err = ProcessConfigFile(conf)
-	if err != nil {
-		t.Fatalf("Error processing config file: %v", err)
-	}
-	if _, err := NewServer(o); err == nil || err.Error() != errMQTTStandaloneNeedsJetStream.Error() {
-		t.Fatalf(`Expected errMQTTStandaloneNeedsJetStream ("next in line"), got %v`, err)
-	}
 }
 
 func TestMQTTStandaloneRequiresJetStream(t *testing.T) {