[FIXED] Stream recovery with corrupt msg block with sequence gaps. (#4344)

This is a fix for a bad msg blk detected in the field that had sequence
holes.

The stream had max msgs per subject of one and only one subject but had
lots of messages. The stream did not recover correctly, and upon further
inspection determined that a msg blk had holes, which should not be
possible.

We now detect the holes and deal with the situation appropriately.
Heavily tested on the data dump from the field.

Signed-off-by: Derek Collison <derek@nats.io>

server/filestore.go

@@ -275,6 +275,8 @@ const (
 	wiThresh = int64(30 * time.Second)
 	// Time threshold to write index info for non FIFO cases
 	winfThresh = int64(2 * time.Second)
+	// Checksum size for hash for msg records.
+	recordHashSize = 8
 )
 
 func newFileStore(fcfg FileStoreConfig, cfg StreamConfig) (*fileStore, error) {
@@ -971,6 +973,10 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 	startLastSeq := mb.last.seq
 
+	// Remove the .fss file and clear any cache we have set.
+	mb.clearCacheAndOffset()
+	mb.removePerSubjectInfoLocked()
+
 	buf, err := mb.loadBlock(nil)
 	if err != nil || len(buf) == 0 {
 		var ld *LostStreamData
@@ -996,9 +1002,6 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 	mb.last.seq, mb.last.ts = 0, 0
 	firstNeedsSet := true
 
-	// Remove the .fss file from disk.
-	mb.removePerSubjectInfoLocked()
-
 	// Check if we need to decrypt.
 	if mb.bek != nil && len(buf) > 0 {
 		// Recreate to reset counter.
@@ -1070,12 +1073,7 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 		rl &^= hbit
 		dlen := int(rl) - msgHdrSize
 		// Do some quick sanity checks here.
-		if dlen < 0 || int(slen) > (dlen-8) || dlen > int(rl) || rl > rlBadThresh {
-			truncate(index)
-			return gatherLost(lbuf - index), errBadMsg
-		}
-
-		if index+rl > lbuf {
+		if dlen < 0 || int(slen) > (dlen-recordHashSize) || dlen > int(rl) || index+rl > lbuf || rl > rlBadThresh {
 			truncate(index)
 			return gatherLost(lbuf - index), errBadMsg
 		}
@@ -1091,15 +1089,17 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 				addToDmap(seq)
 			}
 			index += rl
-			mb.last.seq = seq
-			mb.last.ts = ts
+			if seq >= mb.first.seq {
+				mb.last.seq = seq
+				mb.last.ts = ts
+			}
 			continue
 		}
 
 		// This is for when we have index info that adjusts for deleted messages
 		// at the head. So the first.seq will be already set here. If this is larger
 		// replace what we have with this seq.
-		if firstNeedsSet && seq > mb.first.seq {
+		if firstNeedsSet && seq >= mb.first.seq {
 			firstNeedsSet, mb.first.seq, mb.first.ts = false, seq, ts
 		}
 
@@ -1119,12 +1119,12 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 				hh.Write(hdr[4:20])
 				hh.Write(data[:slen])
 				if hasHeaders {
-					hh.Write(data[slen+4 : dlen-8])
+					hh.Write(data[slen+4 : dlen-recordHashSize])
 				} else {
-					hh.Write(data[slen : dlen-8])
+					hh.Write(data[slen : dlen-recordHashSize])
 				}
 				checksum := hh.Sum(nil)
-				if !bytes.Equal(checksum, data[len(data)-8:]) {
+				if !bytes.Equal(checksum, data[len(data)-recordHashSize:]) {
 					truncate(index)
 					return gatherLost(lbuf - index), errBadMsg
 				}
@@ -1165,6 +1165,11 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 		mb.last.seq = mb.first.seq - 1
 	}
 
+	// Update our fss file if needed.
+	if len(mb.fss) > 0 {
+		mb.writePerSubjectInfo()
+	}
+
 	return nil, nil
 }
 
@@ -2598,14 +2603,42 @@ func (fs *fileStore) enforceMsgPerSubjectLimit() {
 	fs.scb = nil
 	defer func() { fs.scb = cb }()
 
+	var numMsgs uint64
+
 	// collect all that are not correct.
 	needAttention := make(map[string]*psi)
 	for subj, psi := range fs.psim {
+		numMsgs += psi.total
 		if psi.total > maxMsgsPer {
 			needAttention[subj] = psi
 		}
 	}
 
+	// We had an issue with a use case where psim (and hence fss) were correct but idx was not and was not properly being caught.
+	// So do a quick sanity check here. If we detect a skew do a rebuild then re-check.
+	if numMsgs != fs.state.Msgs {
+		// Clear any global subject state.
+		fs.psim = make(map[string]*psi)
+		for _, mb := range fs.blks {
+			mb.removeIndexFile()
+			ld, err := mb.rebuildState()
+			mb.writeIndexInfo()
+			if err != nil && ld != nil {
+				fs.addLostData(ld)
+			}
+			fs.populateGlobalPerSubjectInfo(mb)
+		}
+		// Rebuild fs state too.
+		fs.rebuildStateLocked(nil)
+		// Need to redo blocks that need attention.
+		needAttention = make(map[string]*psi)
+		for subj, psi := range fs.psim {
+			if psi.total > maxMsgsPer {
+				needAttention[subj] = psi
+			}
+		}
+	}
+
 	// Collect all the msgBlks we alter.
 	blks := make(map[*msgBlock]struct{})
 
@@ -3050,8 +3083,7 @@ func (mb *msgBlock) compact() {
 		return
 	}
 
-	// Close cache and index file and wipe delete map, then rebuild.
-	mb.clearCacheAndOffset()
+	// Remove index file and wipe delete map, then rebuild.
 	mb.removeIndexFileLocked()
 	mb.deleteDmap()
 	mb.rebuildStateLocked()
@@ -3077,6 +3109,11 @@ func (mb *msgBlock) slotInfo(slot int) (uint32, uint32, bool, error) {
 	bi := mb.cache.idx[slot]
 	ri, hashChecked := (bi &^ hbit), (bi&hbit) != 0
 
+	// If this is a deleted slot return here.
+	if bi == dbit {
+		return 0, 0, false, errDeletedMsg
+	}
+
 	// Determine record length
 	var rl uint32
 	if len(mb.cache.idx) > slot+1 {
@@ -4022,7 +4059,7 @@ func (fs *fileStore) selectMsgBlockForStart(minTime time.Time) *msgBlock {
 func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 	var le = binary.LittleEndian
 
-	var fseq uint64
+	var fseq, pseq uint64
 	var idx []uint32
 	var index uint32
 
@@ -4055,7 +4092,7 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 		dlen := int(rl) - msgHdrSize
 
 		// Do some quick sanity checks here.
-		if dlen < 0 || int(slen) > dlen || dlen > int(rl) || index+rl > lbuf || rl > 32*1024*1024 {
+		if dlen < 0 || int(slen) > (dlen-recordHashSize) || dlen > int(rl) || index+rl > lbuf || rl > rlBadThresh {
 			// This means something is off.
 			// TODO(dlc) - Add into bad list?
 			return errCorruptState
@@ -4063,15 +4100,31 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 
 		// Clear erase bit.
 		seq = seq &^ ebit
-		// Adjust if we guessed wrong.
-		if seq != 0 && seq < fseq {
-			fseq = seq
-		}
+
 		// We defer checksum checks to individual msg cache lookups to amortorize costs and
 		// not introduce latency for first message from a newly loaded block.
-		idx = append(idx, index)
-		mb.cache.lrl = uint32(rl)
-		index += mb.cache.lrl
+		if seq >= mb.first.seq {
+			// Track that we do not have holes.
+			// Not expected but did see it in the field.
+			if pseq > 0 && seq != pseq+1 {
+				if mb.dmap == nil {
+					mb.dmap = make(map[uint64]struct{})
+				}
+				for dseq := pseq + 1; dseq < seq; dseq++ {
+					idx = append(idx, dbit)
+					mb.dmap[dseq] = struct{}{}
+				}
+			}
+			pseq = seq
+
+			idx = append(idx, index)
+			mb.cache.lrl = uint32(rl)
+			// Adjust if we guessed wrong.
+			if seq != 0 && seq < fseq {
+				fseq = seq
+			}
+		}
+		index += rl
 	}
 	mb.cache.buf = buf
 	mb.cache.idx = idx
@@ -4407,6 +4460,9 @@ const hbit = 1 << 31
 // Used for marking erased messages sequences.
 const ebit = 1 << 63
 
+// Used to mark a bad index as deleted.
+const dbit = 1 << 30
+
 // Will do a lookup from cache.
 // Lock should be held.
 func (mb *msgBlock) cacheLookup(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
@@ -4417,6 +4473,7 @@ func (mb *msgBlock) cacheLookup(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 	// If we have a delete map check it.
 	if mb.dmap != nil {
 		if _, ok := mb.dmap[seq]; ok {
+			mb.llts = time.Now().UnixNano()
 			return nil, errDeletedMsg
 		}
 	}
@@ -4559,9 +4616,9 @@ func (mb *msgBlock) msgFromBuf(buf []byte, sm *StoreMsg, hh hash.Hash64) (*Store
 		hh.Write(hdr[4:20])
 		hh.Write(data[:slen])
 		if hasHeaders {
-			hh.Write(data[slen+4 : dlen-8])
+			hh.Write(data[slen+4 : dlen-recordHashSize])
 		} else {
-			hh.Write(data[slen : dlen-8])
+			hh.Write(data[slen : dlen-recordHashSize])
 		}
 		if !bytes.Equal(hh.Sum(nil), data[len(data)-8:]) {
 			return nil, errBadMsg