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

Issue:
KV with ttl not expired when due
## Defect
we just upgraded to 2.9.7 in our test env.
We have a R3 kv bucket with TTL=5sec being used to campaign for leadership 

```
nats kv info ELECTIONS

Information for Key-Value Store Bucket ELECTIONS created 2022-11-15T17:24:47+01:00

Configuration:

          Bucket Name: ELECTIONS
         History Kept: 1
        Values Stored: 1
   Backing Store Kind: JetStream
          Bucket Size: 133 B
  Maximum Bucket Size: unlimited
   Maximum Value Size: unlimited
     JetStream Stream: KV_ELECTIONS
              Storage: File

  Cluster Information:

                Name: fineco_stage
              Leader: vxr-dev-nats03
             Replica: vxr-dev-nats01, current, seen 0.66s ago
             Replica: vxr-dev-nats02, current, seen 0.66s ago

```


Our process can no longer become leader, apparently not all nodes expire the bucket after 5sec :
Querying the bucket with nats-cli sometime shows :

```
[devmercury@vxr-dev-rundecksvn anagrafica]$ date
Fri Nov 18 15:35:48 CET 2022

[devmercury@vxr-dev-rundecksvn anagrafica]$ nats -s nats://vxr-dev-nats02:4222 kv get ELECTIONS offloading_fe
nats: error: nats: key not found
[devmercury@vxr-dev-rundecksvn anagrafica]$ nats -s nats://vxr-dev-nats02:4222 kv get ELECTIONS offloading_fe
nats: error: nats: key not found
[devmercury@vxr-dev-rundecksvn anagrafica]$ nats -s nats://vxr-dev-nats02:4222 kv get ELECTIONS offloading_fe
nats: error: nats: key not found
[devmercury@vxr-dev-rundecksvn anagrafica]$ nats -s nats://vxr-dev-nats02:4222 kv get ELECTIONS offloading_fe
ELECTIONS > offloading_fe created @ 18 Nov 22 13:57 UTC

```
It is as if the value is kept in a node.

```
nats s info KV_ELECTIONS :
Information for Stream KV_ELECTIONS created 2022-11-15 17:24:47
             Subjects: $KV.ELECTIONS.>
             Replicas: 3
              Storage: File
Options:
            Retention: Limits
     Acknowledgements: true
       Discard Policy: New
     Duplicate Window: 5s
           Direct Get: true
    Allows Msg Delete: false
         Allows Purge: true
       Allows Rollups: true
Limits:
     Maximum Messages: unlimited
  Maximum Per Subject: 1
        Maximum Bytes: unlimited
          Maximum Age: 5.00s
 Maximum Message Size: unlimited
    Maximum Consumers: unlimited
Cluster Information:
                 Name: fineco_stage
               Leader: vxr-dev-nats03
              Replica: vxr-dev-nats01, current, seen 0.81s ago
              Replica: vxr-dev-nats02, current, seen 0.81s ago
State:
             Messages: 1
                Bytes: 133 B
             FirstSeq: 40,565 @ 0001-01-01T00:00:00 UTC
              LastSeq: 66,746 @ 2022-11-18T13:57:16 UTC
     Deleted Messages: 26,181
     Active Consumers: 0
   Number of Subjects: 1
```

Make sure that these boxes are checked before submitting your issue -- thank you!

 - [ ] Included `nats-server -DV` output
 nats-server -DV
[4137868] 2022/11/18 15:54:57.119578 [INF] Starting nats-server
[4137868] 2022/11/18 15:54:57.119649 [INF]   Version:  2.9.7
[4137868] 2022/11/18 15:54:57.119665 [INF]   Git:      [1e76678]
[4137868] 2022/11/18 15:54:57.119675 [DBG]   Go build: go1.19.3
[4137868] 2022/11/18 15:54:57.119680 [INF]   Name:     NC6LMGGJZMNTOFTBXPYG4K6B27O7SXX2FJUW2QUZNGVDJAKUANYKU2PY
[4137868] 2022/11/18 15:54:57.119683 [INF]   ID:       NC6LMGGJZMNTOFTBXPYG4K6B27O7SXX2FJUW2QUZNGVDJAKUANYKU2PY
[4137868] 2022/11/18 15:54:57.119706 [DBG] Created system account: "$SYS"
[4137868] 2022/11/18 15:54:57.121027 [FTL] Error listening on port: 0.0.0.0:4222, "listen tcp 0.0.0.0:4222: bind: address already in use"


 - [ ] Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

I'm keeping the kv with this issue unused let me know if you want access to it

#### Versions of `nats-server` and affected client libraries used:
2.9.7 

This has been opened after a brief talk on slack:  https://natsio.slack.com/archives/CM3T6T7JQ/p1668782422524019



Pull request description:
[FIXED] Bad first sequence and expiration stopped working after server restart.
Fix for the condition after restart where first sequence was wrong and reported zero timestamp and expiration stopped working.

Signed-off-by: Derek Collison <derek@nats.io>

Resolves #3653 

/cc @nats-io/core


Patch:
diff --git a/server/filestore.go b/server/filestore.go
index dfc95388..c62fc6e3 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1205,6 +1205,21 @@ func (fs *fileStore) expireMsgsOnRecover() {
 	var deleted int
 	var nts int64
 
+	deleteEmptyBlock := func(mb *msgBlock) bool {
+		// If we are the last keep state to remember first sequence.
+		if mb == fs.lmb {
+			// Do this part by hand since not deleting one by one.
+			mb.first.seq, mb.first.ts = mb.last.seq+1, 0
+			mb.closeAndKeepIndex()
+			// Clear any global subject state.
+			fs.psim = make(map[string]*psi)
+			return false
+		}
+		mb.dirtyCloseWithRemove(true)
+		deleted++
+		return true
+	}
+
 	for _, mb := range fs.blks {
 		mb.mu.Lock()
 		if minAge < mb.first.ts {
@@ -1216,22 +1231,11 @@ func (fs *fileStore) expireMsgsOnRecover() {
 		if mb.last.ts <= minAge {
 			purged += mb.msgs
 			bytes += mb.bytes
-			// If we are the last keep state to remember first sequence.
-			if mb == fs.lmb {
-				// Do this part by hand since not deleting one by one.
-				mb.first.seq, mb.first.ts = mb.last.seq+1, 0
-				mb.closeAndKeepIndex()
-				// Clear any global subject state.
-				fs.psim = make(map[string]*psi)
-			} else {
-				mb.dirtyCloseWithRemove(true)
-				deleted++
-			}
-			newFirst := mb.last.seq + 1
+			didRemove := deleteEmptyBlock(mb)
 			mb.mu.Unlock()
-			// Update fs first here as well.
-			fs.state.FirstSeq = newFirst
-			fs.state.FirstTime = time.Time{}
+			if !didRemove {
+				mb.writeIndexInfo()
+			}
 			continue
 		}
 
@@ -1292,22 +1296,12 @@ func (fs *fileStore) expireMsgsOnRecover() {
 		}
 
 		// Check if empty after processing, could happen if tail of messages are all deleted.
-		isEmpty := mb.msgs == 0
-		if isEmpty {
-			mb.dirtyCloseWithRemove(true)
-			// Update fs first here as well.
-			fs.state.FirstSeq = mb.last.seq + 1
-			fs.state.FirstTime = time.Time{}
-			deleted++
-		} else {
-			// Update fs first seq and time.
-			fs.state.FirstSeq = mb.first.seq
-			fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
+		needWriteIndex := true
+		if mb.msgs == 0 {
+			needWriteIndex = !deleteEmptyBlock(mb)
 		}
 		mb.mu.Unlock()
-
-		if !isEmpty {
-			// Make sure to write out our index info.
+		if needWriteIndex {
 			mb.writeIndexInfo()
 		}
 		break
@@ -1336,6 +1330,8 @@ func (fs *fileStore) expireMsgsOnRecover() {
 	// Update top level accounting.
 	fs.state.Msgs -= purged
 	fs.state.Bytes -= bytes
+	// Make sure to we properly set the fs first sequence and timestamp.
+	fs.selectNextFirst()
 }
 
 func copyMsgBlocks(src []*msgBlock) []*msgBlock {
@@ -2884,6 +2880,7 @@ func (mb *msgBlock) selectNextFirst() {
 }
 
 // Select the next FirstSeq
+// Lock should be held.
 func (fs *fileStore) selectNextFirst() {
 	if len(fs.blks) > 0 {
 		mb := fs.blks[0]
@@ -3054,8 +3051,12 @@ func (fs *fileStore) startAgeChk() {
 
 // Lock should be held.
 func (fs *fileStore) resetAgeChk(delta int64) {
+	if fs.cfg.MaxAge == 0 {
+		return
+	}
+
 	fireIn := fs.cfg.MaxAge
-	if delta > 0 {
+	if delta > 0 && time.Duration(delta) < fireIn {
 		fireIn = time.Duration(delta)
 	}
 	if fs.ageChk != nil {
@@ -3089,10 +3090,15 @@ func (fs *fileStore) expireMsgs() {
 	fs.mu.Lock()
 	defer fs.mu.Unlock()
 
-	if sm == nil {
+	// Onky cancel if no message left, not on potential lookup error that would result in sm == nil.
+	if fs.state.Msgs == 0 {
 		fs.cancelAgeChk()
 	} else {
-		fs.resetAgeChk(sm.ts - minAge)
+		if sm == nil {
+			fs.resetAgeChk(0)
+		} else {
+			fs.resetAgeChk(sm.ts - minAge)
+		}
 	}
 }
 


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 89ed90e1..107779aa 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -4764,3 +4764,72 @@ func TestFileStoreUpdateMaxMsgsPerSubject(t *testing.T) {
 		t.Fatalf("Expected to have %d stored, got %d", 10, ss.Msgs)
 	}
 }
+
+func TestFileStoreBadFirstAndFailedExpireAfterRestart(t *testing.T) {
+	storeDir := t.TempDir()
+
+	ttl := time.Second
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 256},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage, MaxAge: ttl},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// With block size of 256 and subject and message below, seq 8 starts new block.
+	// Will double check and fail test if not the case since test depends on this.
+	subj, msg := "foo", []byte("ZZ")
+	// These are all instant and will expire after 1 sec.
+	start := time.Now()
+	for i := 0; i < 7; i++ {
+		_, _, err := fs.StoreMsg(subj, nil, msg)
+		require_NoError(t, err)
+	}
+
+	// Put two more after a delay.
+	time.Sleep(500 * time.Millisecond)
+	seq, _, err := fs.StoreMsg(subj, nil, msg)
+	require_NoError(t, err)
+	_, _, err = fs.StoreMsg(subj, nil, msg)
+	require_NoError(t, err)
+
+	// Make sure that sequence 8 is first in second block, and break test if that is not true.
+	fs.mu.RLock()
+	lmb := fs.lmb
+	fs.mu.RUnlock()
+	lmb.mu.RLock()
+	first := lmb.first.seq
+	lmb.mu.RUnlock()
+	require_True(t, first == 8)
+
+	// Instantly remove first one from second block.
+	// On restart this will trigger expire on recover which will set fs.FirstSeq to the deleted one.
+	fs.RemoveMsg(seq)
+	// Stop the filstore and wait til first block expires.
+	fs.Stop()
+	time.Sleep(ttl - time.Since(start) + (10 * time.Millisecond))
+
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 256},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage, MaxAge: ttl},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// Check that state is correct for first message which should be 9 and have a proper timestamp.
+	var state StreamState
+	fs.FastState(&state)
+	ts := state.FirstTime
+	require_True(t, state.Msgs == 1)
+	require_True(t, state.FirstSeq == 9)
+	require_True(t, !state.FirstTime.IsZero())
+
+	// Wait and make sure expire timer is still working properly.
+	time.Sleep(ttl)
+	fs.FastState(&state)
+	require_True(t, state.Msgs == 0)
+	require_True(t, state.FirstSeq == 10)
+	require_True(t, state.LastSeq == 9)
+	require_True(t, state.LastTime == ts)
+}