Repository: nats-io/nats-server
Pull request #2859 (2022-02-10 00:26:57)

Remove fss files from a snapshot when a block is removed.
During a filestore snapshot we generate the fss files for the snapshot but were not cleaning them up if the block was deleted before a server restart.

https://gist.github.com/nekufa/010185dfb59261f222a0042d3a7d2a1c

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 0e1ddc4d..aee81c0a 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -239,7 +239,6 @@ type consumer struct {
 	maxdc             uint64
 	waiting           *waitQueue
 	cfg               ConsumerConfig
-	ici               *ConsumerInfo
 	store             ConsumerStore
 	active            bool
 	replay            bool
@@ -850,9 +849,6 @@ func (o *consumer) setLeader(isLeader bool) {
 		}
 		o.mu.Unlock()
 
-		// Snapshot initial info.
-		o.infoWithSnap(true)
-
 		// Now start up Go routine to deliver msgs.
 		go o.loopAndGatherMsgs(qch)
 
@@ -1834,33 +1830,8 @@ func (o *consumer) writeStoreStateUnlocked() error {
 	return o.store.Update(&state)
 }
 
-// Returns an initial info. Only applicable for non-clustered consumers.
-// We will clear after we return it, so one shot.
-func (o *consumer) initialInfo() *ConsumerInfo {
-	o.mu.Lock()
-	ici := o.ici
-	o.ici = nil // gc friendly
-	o.mu.Unlock()
-	if ici == nil {
-		ici = o.info()
-	}
-	return ici
-}
-
-// Clears our initial info.
-// Used when we have a leader change in cluster mode but do not send a response.
-func (o *consumer) clearInitialInfo() {
-	o.mu.Lock()
-	o.ici = nil // gc friendly
-	o.mu.Unlock()
-}
-
 // Info returns our current consumer state.
 func (o *consumer) info() *ConsumerInfo {
-	return o.infoWithSnap(false)
-}
-
-func (o *consumer) infoWithSnap(snap bool) *ConsumerInfo {
 	o.mu.RLock()
 	mset := o.mset
 	if mset == nil || mset.srv == nil {
@@ -1914,11 +1885,6 @@ func (o *consumer) infoWithSnap(snap bool) *ConsumerInfo {
 		o.expireWaiting()
 		info.NumWaiting = o.waiting.len()
 	}
-	// If we were asked to snapshot do so here.
-	if snap {
-		o.ici = info
-	}
-
 	return info
 }
 
diff --git a/server/filestore.go b/server/filestore.go
index 055a8dea..04382869 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -30,6 +30,7 @@ import (
 	"net"
 	"os"
 	"path"
+	"path/filepath"
 	"runtime"
 	"sort"
 	"sync"
@@ -187,6 +188,8 @@ const (
 	indexScan = "%d.idx"
 	// used to load per subject meta information.
 	fssScan = "%d.fss"
+	// to look for orphans
+	fssScanAll = "*.fss"
 	// used to store our block encryption key.
 	keyScan = "%d.key"
 	// This is where we keep state on consumers.
@@ -890,7 +893,6 @@ func (fs *fileStore) recoverMsgs() error {
 						fs.psmc[subj] += ss.Msgs
 					}
 				}
-
 			} else {
 				return err
 			}
@@ -909,6 +911,14 @@ func (fs *fileStore) recoverMsgs() error {
 		return err
 	}
 
+	// We had a bug that would leave fss files around during a snapshot.
+	// Clean them up here if we see them.
+	if fms, err := filepath.Glob(path.Join(mdir, fssScanAll)); err == nil && len(fms) > 0 {
+		for _, fn := range fms {
+			os.Remove(fn)
+		}
+	}
+
 	// Limits checks and enforcement.
 	fs.enforceMsgLimit()
 	fs.enforceBytesLimit()
@@ -4326,6 +4336,10 @@ func (mb *msgBlock) dirtyCloseWithRemove(remove bool) {
 			os.Remove(mb.mfn)
 			mb.mfn = _EMPTY_
 		}
+		if mb.sfn != _EMPTY_ {
+			os.Remove(mb.sfn)
+			mb.sfn = _EMPTY_
+		}
 	}
 }
 
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index f01651cb..fde67a31 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -3244,7 +3244,7 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subj
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
-	resp.ConsumerInfo = o.initialInfo()
+	resp.ConsumerInfo = o.info()
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index f5fa25c8..4ceb97ae 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -3292,9 +3292,6 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) {
 	}
 
 	if !isLeader || hasResponded {
-		if isLeader {
-			o.clearInitialInfo()
-		}
 		return
 	}
 
@@ -3303,7 +3300,7 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) {
 		resp.Error = NewJSConsumerCreateError(err, Unless(err))
 		s.sendAPIErrResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
 	} else {
-		resp.ConsumerInfo = o.initialInfo()
+		resp.ConsumerInfo = o.info()
 		s.sendAPIResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
 		if node := o.raftNode(); node != nil {
 			o.sendCreateAdvisory()


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 3082b2b8..93618f36 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -10457,56 +10457,6 @@ func TestJetStreamConsumerUpgrade(t *testing.T) {
 	t.Run("Clustered", func(t *testing.T) { testUpdate(t, c.randomServer()) })
 }
 
-func TestJetStreamAddConsumerWithInfo(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer s.Shutdown()
-
-	if config := s.JetStreamConfig(); config != nil {
-		defer removeDir(t, config.StoreDir)
-	}
-
-	c := createJetStreamClusterExplicit(t, "JSC", 3)
-	defer c.shutdown()
-
-	testConsInfo := func(t *testing.T, s *Server) {
-		nc, js := jsClientConnect(t, s)
-		defer nc.Close()
-
-		_, err := js.AddStream(&nats.StreamConfig{
-			Name:     "TEST",
-			Subjects: []string{"foo"},
-		})
-		require_NoError(t, err)
-
-		for i := 0; i < 10; i++ {
-			_, err = js.Publish("foo", []byte("msg"))
-			require_NoError(t, err)
-		}
-
-		for i := 0; i < 100; i++ {
-			inbox := nats.NewInbox()
-			sub := natsSubSync(t, nc, inbox)
-
-			ci, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
-				DeliverSubject: inbox,
-				DeliverPolicy:  nats.DeliverAllPolicy,
-				FilterSubject:  "foo",
-				AckPolicy:      nats.AckExplicitPolicy,
-			})
-			require_NoError(t, err)
-
-			if ci.NumPending != 10 {
-				t.Fatalf("Iter=%v - expected 10 messages pending on create, got %v", i+1, ci.NumPending)
-			}
-			js.DeleteConsumer("TEST", ci.Name)
-			sub.Unsubscribe()
-		}
-	}
-
-	t.Run("Single", func(t *testing.T) { testConsInfo(t, s) })
-	t.Run("Clustered", func(t *testing.T) { testConsInfo(t, c.randomServer()) })
-}
-
 // Support functions
 
 // Used to setup superclusters for tests.
diff --git a/server/norace_test.go b/server/norace_test.go
index 968f582c..80164b76 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -29,6 +29,8 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"path"
+	"path/filepath"
 	"runtime"
 	"runtime/debug"
 	"strconv"
@@ -4275,3 +4277,50 @@ func TestNoRaceJetStreamSparseConsumers(t *testing.T) {
 		})
 	}
 }
+
+func TestNoRaceFileStoreSubjectInfoWithSnapshotCleanup(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 1024 * 1024}, StreamConfig{Name: "TEST", Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	n, msg := 10_000, []byte(strings.Repeat("Z", 1024))
+	for i := 0; i < n; i++ {
+		_, _, err := fs.StoreMsg(fmt.Sprintf("X.%d", i), nil, msg)
+		require_NoError(t, err)
+	}
+
+	// Snapshot causes us to write out per subject info, fss files.
+	// We want to make sure they get cleaned up.
+	sr, err := fs.Snapshot(5*time.Second, false, false)
+	require_NoError(t, err)
+	var buf [4 * 1024 * 1024]byte
+	for {
+		if _, err = sr.Reader.Read(buf[:]); err == io.EOF {
+			break
+		}
+		require_NoError(t, err)
+	}
+
+	var seqs []uint64
+	for i := 1; i <= n; i++ {
+		seqs = append(seqs, uint64(i))
+	}
+	// Randomly delete msgs, make sure we cleanup as we empty the message blocks.
+	rand.Shuffle(len(seqs), func(i, j int) { seqs[i], seqs[j] = seqs[j], seqs[i] })
+
+	for _, seq := range seqs {
+		_, err := fs.RemoveMsg(seq)
+		require_NoError(t, err)
+	}
+
+	// We will have cleanup the main .blk and .idx sans the lmb, but we should not have any *.fss files.
+	fms, err := filepath.Glob(path.Join(storeDir, msgDir, fssScanAll))
+	require_NoError(t, err)
+
+	if len(fms) > 0 {
+		t.Fatalf("Expected to find no fss files, found %d", len(fms))
+	}
+}