Repository: nats-io/nats-server
Pull request #3284 (2022-07-24 00:12:21)

[FIXED] Losing stream sequence on multiple restarts and leader changes.
When msgs were expired on restart recovery we could lose track on subsequent restarts of our starting sequence with no additional activity. If a leadership change happened or we were in R1 mode we would lose our starting sequence.

This could happen on low traffic streams with TTLs that frequently would empty the stream and when a server was down possibly before that event happened.

Thanks to @tbeets for the details on how to reproduce!

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 77a6a2f5..ee71440b 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1067,13 +1067,20 @@ func (fs *fileStore) expireMsgsOnRecover() {
 		if mb.last.ts <= minAge {
 			purged += mb.msgs
 			bytes += mb.bytes
-			mb.dirtyCloseWithRemove(true)
+			// 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()
+			} else {
+				mb.dirtyCloseWithRemove(true)
+				deleted++
+			}
 			newFirst := mb.last.seq + 1
 			mb.mu.Unlock()
 			// Update fs first here as well.
 			fs.state.FirstSeq = newFirst
 			fs.state.FirstTime = time.Time{}
-			deleted++
 			continue
 		}
 


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 6b905312..95906717 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3102,6 +3102,7 @@ func TestFileStoreExpireMsgsOnStart(t *testing.T) {
 	}
 
 	checkNumBlks := func(expected int) {
+		t.Helper()
 		fs.mu.RLock()
 		n := len(fs.blks)
 		fs.mu.RUnlock()
@@ -3174,7 +3175,8 @@ func TestFileStoreExpireMsgsOnStart(t *testing.T) {
 	loadMsgs(500)
 	restartFS(ttl + 100*time.Millisecond)
 	checkState(0, 501, 500)
-	checkNumBlks(0)
+	// We actually hold onto the last one now to remember our starting sequence.
+	checkNumBlks(1)
 
 	// Now check partial expires and the fss tracking state.
 	// Small numbers is to keep them in one block.
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 37eaf25f..f49b7d15 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -11979,3 +11979,86 @@ func TestJetStreamClusterNoOrphanedDueToNoConnection(t *testing.T) {
 	time.Sleep(7 * eventsHBInterval)
 	checkSysServers()
 }
+
+func TestJetStreamClusterStreamResetOnExpirationDuringPeerDownAndRestartWithLeaderChange(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3F", 3)
+	defer c.shutdown()
+
+	s := c.randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+		MaxAge:   time.Second,
+	})
+	require_NoError(t, err)
+
+	n := 100
+	for i := 0; i < n; i++ {
+		js.PublishAsync("foo", []byte("NORESETPLS"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	// Shutdown a non-leader before expiration.
+	nsl := c.randomNonStreamLeader("$G", "TEST")
+	nsl.Shutdown()
+
+	// Wait for all messages to expire.
+	checkFor(t, 2*time.Second, 20*time.Millisecond, func() error {
+		si, err := js.StreamInfo("TEST")
+		require_NoError(t, err)
+		if si.State.Msgs == 0 {
+			return nil
+		}
+		return fmt.Errorf("Wanted 0 messages, got %d", si.State.Msgs)
+	})
+
+	// Now restart the non-leader server, twice. First time clears raft,
+	// second will not have any index state or raft to tell it what is first sequence.
+	nsl = c.restartServer(nsl)
+	c.checkClusterFormed()
+	c.waitOnServerCurrent(nsl)
+
+	// Now clear raft WAL.
+	mset, err := nsl.GlobalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	require_NoError(t, mset.raftNode().InstallSnapshot(mset.stateSnapshot()))
+
+	nsl.Shutdown()
+	nsl = c.restartServer(nsl)
+	c.checkClusterFormed()
+	c.waitOnServerCurrent(nsl)
+
+	// We will now check this server directly.
+	mset, err = nsl.GlobalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+
+	if state := mset.state(); state.FirstSeq != uint64(n+1) {
+		t.Fatalf("Expected first sequence of %d, got %d", n+1, state.FirstSeq)
+	}
+
+	// Now move the leader there and double check, but above test is sufficient.
+	checkFor(t, 10*time.Second, 250*time.Millisecond, func() error {
+		_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "TEST"), nil, time.Second)
+		require_NoError(t, err)
+		c.waitOnStreamLeader("$G", "TEST")
+		if c.streamLeader("$G", "TEST") == nsl {
+			return nil
+		}
+		return fmt.Errorf("No correct leader yet")
+	})
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+
+	if state := si.State; state.FirstSeq != uint64(n+1) {
+		t.Fatalf("Expected first sequence of %d, got %d", n+1, state.FirstSeq)
+	}
+}