Repository: nats-io/nats-server (go)

Issue:
New keys with TTL not deleted after expiration following update to include marker TTL
### Observed behavior

I encountered an issue with the key-value store involving marker TTL functionality. Initially, I created the key-value store without setting a marker TTL. Later, I edited and updated the store to include a marker TTL and then added a new key with its own TTL. However, I've observed that the entity is not automatically deleted once its TTL expires.

### Expected behavior

Upon editing the key-value store to include a marker TTL, any subsequently added key with its own TTL should be automatically deleted once the TTL expires.

Is this the behavior that is expected?

### Server and client version

Server version: 2.11.3
Client version: 0.2.3

### Host environment

Windows and Kubernetes

### Steps to reproduce

```bash
nats kv add testkv
nats kv edit testkv --marker-ttl=30s
nats kv create testkv key value --ttl=20s
#Sleep
sleep 25s
nats kv get testkv key
#expectation is to not find the key. But the key is found
```

Pull request description:
Correctly handle updates to `AllowMsgTTL` on streams
Fixes #6920.

Signed-off-by: Neil Twigg <neil@nats.io>

Patch:
diff --git a/.github/ISSUE_TEMPLATE/defect.yml b/.github/ISSUE_TEMPLATE/defect.yml
index 4bdb0e81..4816b811 100644
--- a/.github/ISSUE_TEMPLATE/defect.yml
+++ b/.github/ISSUE_TEMPLATE/defect.yml
@@ -23,7 +23,7 @@ body:
     attributes:
       label: Server and client version
       description: |-
-        **One of the first things we are likely to ask you is if the defect reproduces on the latest version, so please check that first!**
+        Provide the versions you were using when the detect was observed.
         For the server, use `nats-server --version`, check the startup log output, or the image tag pulled from Docker.
         For the CLI client, use `nats --version`.
         For language-specific clients, check the version downloaded by the language dependency manager.
@@ -42,6 +42,6 @@ body:
     id: steps
     attributes:
       label: Steps to reproduce
-      description: Provide as many concrete steps to reproduce the defect. Please also include stream info and consumer info if this issue is JetStream-related.
+      description: Provide as many concrete steps to reproduce the defect.
     validations:
       required: false
diff --git a/server/filestore.go b/server/filestore.go
index 392e7135..5ad6442a 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -648,6 +648,13 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 		return err
 	}
 
+	// Create or delete the THW if needed.
+	if cfg.AllowMsgTTL && fs.ttls == nil {
+		fs.ttls = thw.NewHashWheel()
+	} else if !cfg.AllowMsgTTL && fs.ttls != nil {
+		fs.ttls = nil
+	}
+
 	// Limits checks and enforcement.
 	fs.enforceMsgLimit()
 	fs.enforceBytesLimit()
@@ -666,7 +673,7 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 	}
 	fs.mu.Unlock()
 
-	if cfg.MaxAge != 0 {
+	if cfg.MaxAge != 0 || cfg.AllowMsgTTL {
 		fs.expireMsgs()
 	}
 	return nil
diff --git a/server/memstore.go b/server/memstore.go
index 4da9adff..581edb5e 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -83,6 +83,12 @@ func (ms *memStore) UpdateConfig(cfg *StreamConfig) error {
 
 	ms.mu.Lock()
 	ms.cfg = *cfg
+	// Create or delete the THW if needed.
+	if cfg.AllowMsgTTL && ms.ttls == nil {
+		ms.ttls = thw.NewHashWheel()
+	} else if !cfg.AllowMsgTTL && ms.ttls != nil {
+		ms.ttls = nil
+	}
 	// Limits checks and enforcement.
 	ms.enforceMsgLimit()
 	ms.enforceBytesLimit()
@@ -112,7 +118,7 @@ func (ms *memStore) UpdateConfig(cfg *StreamConfig) error {
 	}
 	ms.mu.Unlock()
 
-	if cfg.MaxAge != 0 {
+	if cfg.MaxAge != 0 || cfg.AllowMsgTTL {
 		ms.expireMsgs()
 	}
 	return nil


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 78ccf486..22592e3f 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -9607,3 +9607,23 @@ func TestFileStoreAccessTimeSpinUp(t *testing.T) {
 	ngra := runtime.NumGoroutine()
 	require_Equal(t, ngr, ngra)
 }
+
+func TestFileStoreUpdateConfigTTLState(t *testing.T) {
+	cfg := StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{">"},
+		Storage:  FileStorage,
+	}
+	fs, err := newFileStore(FileStoreConfig{StoreDir: t.TempDir()}, cfg)
+	require_NoError(t, err)
+	defer fs.Stop()
+	require_Equal(t, fs.ttls, nil)
+
+	cfg.AllowMsgTTL = true
+	require_NoError(t, fs.UpdateConfig(&cfg))
+	require_NotEqual(t, fs.ttls, nil)
+
+	cfg.AllowMsgTTL = false
+	require_NoError(t, fs.UpdateConfig(&cfg))
+	require_Equal(t, fs.ttls, nil)
+}
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 77ab3a2c..2f95ee3e 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -1292,6 +1292,26 @@ func TestMemStoreAllLastSeqs(t *testing.T) {
 	require_True(t, reflect.DeepEqual(seqs, expected))
 }
 
+func TestMemStoreUpdateConfigTTLState(t *testing.T) {
+	cfg := &StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{">"},
+		Storage:  MemoryStorage,
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(t, err)
+	defer ms.Stop()
+	require_Equal(t, ms.ttls, nil)
+
+	cfg.AllowMsgTTL = true
+	require_NoError(t, ms.UpdateConfig(cfg))
+	require_NotEqual(t, ms.ttls, nil)
+
+	cfg.AllowMsgTTL = false
+	require_NoError(t, ms.UpdateConfig(cfg))
+	require_Equal(t, ms.ttls, nil)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/store_test.go b/server/store_test.go
index 05f09993..32d56748 100644
--- a/server/store_test.go
+++ b/server/store_test.go
@@ -18,6 +18,7 @@ package server
 import (
 	"fmt"
 	"testing"
+	"time"
 )
 
 func testAllStoreAllPermutations(t *testing.T, compressionAndEncryption bool, cfg StreamConfig, fn func(t *testing.T, fs StreamStore)) {
@@ -535,3 +536,50 @@ func TestStorePurgeExZero(t *testing.T) {
 		},
 	)
 }
+
+func TestStoreUpdateConfigTTLState(t *testing.T) {
+	config := func() StreamConfig {
+		return StreamConfig{Name: "TEST", Subjects: []string{"foo"}}
+	}
+	testAllStoreAllPermutations(
+		t, false, config(),
+		func(t *testing.T, fs StreamStore) {
+			cfg := config()
+			switch fs.(type) {
+			case *fileStore:
+				cfg.Storage = FileStorage
+			case *memStore:
+				cfg.Storage = MemoryStorage
+			}
+
+			// TTLs disabled at this point so this message should survive.
+			seq, _, err := fs.StoreMsg("foo", nil, nil, 1)
+			require_NoError(t, err)
+			time.Sleep(2 * time.Second)
+			_, err = fs.LoadMsg(seq, nil)
+			require_NoError(t, err)
+
+			// Now enable TTLs.
+			cfg.AllowMsgTTL = true
+			require_NoError(t, fs.UpdateConfig(&cfg))
+
+			// TTLs enabled at this point so this message should be cleaned up.
+			seq, _, err = fs.StoreMsg("foo", nil, nil, 1)
+			require_NoError(t, err)
+			time.Sleep(2 * time.Second)
+			_, err = fs.LoadMsg(seq, nil)
+			require_Error(t, err)
+
+			// Now disable TTLs again.
+			cfg.AllowMsgTTL = false
+			require_NoError(t, fs.UpdateConfig(&cfg))
+
+			// TTLs disabled again so this message should survive.
+			seq, _, err = fs.StoreMsg("foo", nil, nil, 1)
+			require_NoError(t, err)
+			time.Sleep(2 * time.Second)
+			_, err = fs.LoadMsg(seq, nil)
+			require_NoError(t, err)
+		},
+	)
+}