Repository: nats-io/nats-server
Pull request #1880 (2021-02-03 22:00:42)

[FIXED] Bug with erased messages and re-indexing the message block cache.
We needed to clear the erase bit to properly set fseq in the cache structure on re-indexing.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 7a3eab50..403b1c48 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1010,6 +1010,10 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 	cb := fs.scb
 	mb.mu.Unlock()
 
+	if secure {
+		mb.flushPendingMsgs()
+	}
+
 	// Kick outside of lock.
 	if shouldWriteIndex {
 		if qch == nil {
@@ -1207,6 +1211,7 @@ func (mb *msgBlock) eraseMsg(seq uint64, ri, rl int) error {
 		buf := mb.cache.buf[li : li+rl]
 		copy(buf, nbytes)
 	}
+
 	// Disk
 	if mb.cache.off+mb.cache.wp > ri {
 		mfd, err := os.OpenFile(mb.mfn, os.O_RDWR, 0644)
@@ -1879,6 +1884,8 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 			// TODO(dlc) - Add into bad list?
 			return errBadMsg
 		}
+		// Clear erase bit.
+		seq = seq &^ ebit
 		// Adjust if we guessed wrong.
 		if seq != 0 && seq < fseq {
 			fseq = seq
@@ -2090,6 +2097,7 @@ func (mb *msgBlock) fetchMsg(seq uint64) (*fileStoredMsg, error) {
 	if err := mb.loadMsgs(); err != nil {
 		return nil, err
 	}
+
 	return mb.cacheLookup(seq)
 }
 
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 1cc792f0..971313dd 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -333,7 +333,8 @@ const JSApiStreamUpdateResponseType = "io.nats.jetstream.api.v1.stream_update_re
 
 // JSApiMsgDeleteRequest delete message request.
 type JSApiMsgDeleteRequest struct {
-	Seq uint64 `json:"seq"`
+	Seq     uint64 `json:"seq"`
+	NoErase bool   `json:"no_erase,omitempty"`
 }
 
 // JSApiMsgDeleteResponse.
@@ -1835,7 +1836,12 @@ func (s *Server) jsMsgDeleteRequest(sub *subscription, c *client, subject, reply
 		return
 	}
 
-	removed, err := mset.EraseMsg(req.Seq)
+	var removed bool
+	if req.NoErase {
+		removed, err = mset.RemoveMsg(req.Seq)
+	} else {
+		removed, err = mset.EraseMsg(req.Seq)
+	}
 	if err != nil {
 		resp.Error = jsError(err)
 	} else if !removed {


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index e05f2440..b36389a8 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -2551,3 +2551,29 @@ func TestFileStoreStreamIndexBug(t *testing.T) {
 		t.Fatalf("Expected error during readIndexInfo(): %v", err)
 	}
 }
+
+// Reported by Ivan.
+func TestFileStoreStreamDeleteCacheBug(t *testing.T) {
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: 50 * time.Millisecond}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("Hello World")
+
+	if _, _, err := fs.StoreMsg(subj, nil, msg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if _, _, err := fs.StoreMsg(subj, nil, msg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if _, err := fs.EraseMsg(1); err != nil {
+		t.Fatalf("Got an error on remove of %d: %v", 1, err)
+	}
+	time.Sleep(100 * time.Millisecond)
+	if _, _, _, _, err := fs.LoadMsg(2); err != nil {
+		t.Fatalf("Unexpected error looking up msg: %v", err)
+	}
+}