Add support for "never expires" TTL (#6319)

A message with `Nats-TTL: never` will never expire, even if the `MaxAge`
policy is set.

Signed-off-by: Neil Twigg <[email protected]>

Author: derekcollison

server/filestore.go

@@ -5304,7 +5304,15 @@ func (fs *fileStore) expireMsgs() {
 	fs.mu.RUnlock()
 
 	if maxAge > 0 {
-		for sm, _ = fs.msgForSeq(0, &smv); sm != nil && sm.ts <= minAge; sm, _ = fs.msgForSeq(0, &smv) {
+		var seq uint64
+		for sm, seq, _ = fs.LoadNextMsg(fwcs, true, 0, &smv); sm != nil && sm.ts <= minAge; sm, seq, _ = fs.LoadNextMsg(fwcs, true, seq+1, &smv) {
+			if len(sm.hdr) > 0 {
+				if ttl, err := getMessageTTL(sm.hdr); err == nil && ttl < 0 {
+					// The message has a negative TTL, therefore it must "never expire".
+					minAge = time.Now().UnixNano() - maxAge
+					continue
+				}
+			}
 			fs.mu.Lock()
 			fs.removeMsgViaLimits(sm.seq)
 			fs.mu.Unlock()

server/jetstream_test.go

@@ -25016,3 +25016,51 @@ func TestJetStreamMessageTTLNotUpdatable(t *testing.T) {
 	})
 	require_Error(t, err)
 }
+
+func TestJetStreamMessageTTLNeverExpire(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	jsStreamCreate(t, nc, &StreamConfig{
+		Name:        "TEST",
+		Storage:     FileStorage,
+		Subjects:    []string{"test"},
+		AllowMsgTTL: true,
+		MaxAge:      time.Second,
+	})
+
+	msg := &nats.Msg{
+		Subject: "test",
+		Header:  nats.Header{},
+	}
+
+	// The first message we publish is set to "never expire", therefore it
+	// won't age out with the MaxAge policy.
+	msg.Header.Set("Nats-TTL", "never")
+	_, err := js.PublishMsg(msg)
+	require_NoError(t, err)
+
+	// Following messages will be published as normal and will age out.
+	msg.Header.Del("Nats-TTL")
+	for i := 1; i <= 10; i++ {
+		_, err := js.PublishMsg(msg)
+		require_NoError(t, err)
+	}
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 11)
+	require_Equal(t, si.State.FirstSeq, 1)
+	require_Equal(t, si.State.LastSeq, 11)
+
+	time.Sleep(time.Second * 2)
+
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 1)
+	require_Equal(t, si.State.FirstSeq, 1)
+	require_Equal(t, si.State.LastSeq, 11)
+}

server/stream.go

@@ -4220,12 +4220,16 @@ func getExpectedLastSeqPerSubjectForSubject(hdr []byte) string {
 // Fast lookup of the message TTL:
 // - Positive return value: duration in seconds.
 // - Zero return value: no TTL or parse error.
+// - Negative return value: never expires.
 func getMessageTTL(hdr []byte) (int64, error) {
 	ttl := getHeader(JSMessageTTL, hdr)
 	if len(ttl) == 0 {
 		return 0, nil
 	}
 	sttl := bytesToString(ttl)
+	if strings.ToLower(sttl) == "never" {
+		return -1, nil
+	}
 	dur, err := time.ParseDuration(sttl)
 	if err == nil {
 		if dur < time.Second {
@@ -4235,6 +4239,10 @@ func getMessageTTL(hdr []byte) (int64, error) {
 	}
 	t := parseInt64(ttl)
 	if t < 0 {
+		// This probably means a parse failure, hence why
+		// we have a special case "never" for returning -1.
+		// Otherwise we can't know if it's a genuine TTL
+		// that says never expire or if it's a parse error.
 		return 0, NewJSMessageTTLInvalidError()
 	}
 	return t, nil