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

Issue:
Consumers stops receiving messages
## Defect

#### Versions of `nats-server` and affected client libraries used:

Nats server version
```
[83] 2021/09/04 18:51:12.239432 [INF] Starting nats-server
[83] 2021/09/04 18:51:12.239488 [INF]   Version:  2.4.0
[83] 2021/09/04 18:51:12.239494 [INF]   Git:      [219a7c98]
[83] 2021/09/04 18:51:12.239496 [DBG]   Go build: go1.16.7
[83] 2021/09/04 18:51:12.239517 [INF]   Name:     NBVE7O7DMRAZ63STC7Z644KHF5HJ6QQUGLZVGDIKEG32CFL2J6O2456M
[83] 2021/09/04 18:51:12.239533 [INF]   ID:       NBVE7O7DMRAZ63STC7Z644KHF5HJ6QQUGLZVGDIKEG32CFL2J6O2456M
[83] 2021/09/04 18:51:12.239605 [DBG] Created system account: "$SYS"
```

Go client version: `v1.12.0`

#### OS/Container environment:

GKE Kubernetes. Running nats js HA cluster. Deployed via nats helm chart.

#### Steps or code to reproduce the issue:

Stream configuration:
```yaml
apiVersion: jetstream.nats.io/v1beta1
kind: Stream
metadata:
  name: agent
spec:
  name: agent
  subjects: ["data.*"]
  storage: file
  maxAge: 1h
  replicas: 3
  retention: interest
```

There are two consumers to this stream. Each runs as queue subscriber in two services with 2 pod replicas. Note that I don't care if message is not processed, this is why ack none is set.
```go

// 2 pods for service A.
js.QueueSubscribe(
	"data.received",
	"service1_queue",
	func(msg *nats.Msg) {},
	nats.DeliverNew(),
	nats.AckNone(),
)

// 2 pods for service B.
s.js.QueueSubscribe(
	"data.received",
	"service2_queue",
	func(msg *nats.Msg) {},
	nats.DeliverNew(),
	nats.AckNone(),
)
```

#### Expected result:

Consumer receives messages.

#### Actual result:


Stream stats after few days:

```
agent                  │ File    │ 3         │ 28,258   │ 18 MiB  │ 0    │ 84      │ nats-js-0, nats-js-1*, nats-js-2
```

Consumers stats:

```
service1_queue │ Push │ None       │ 0.00s    │ 0           │ 0           │ 0           │ 60,756    │ nats-js-0, nats-js-1*, nats-js-2
service2_queue │ Push │ None       │ 0.00s    │ 0           │ 0           │ 8,193 / 28% │ 60,843    │ nats-js-0, nats-js-1*, nats-js-2
```

1. Non of the nats server pods contains errors indicating any problem.
2. Unprocessed messages count for second consumer stays the same and doesn't decrease.
3. The only fix which helped is after I changed second consumer raft leader with `nats consumer cluster step-down`. But after some time problem still comes back.
4. There are active connections to the server. Checked with `nats server report connections`.


/cc @kozlovic @derekcollison 

Pull request:
[FIXED] #2488
When we triggered a filestore msg block compact we were not properly dealing with interior deletes.
 Subsequent lookups past the skipped messages would cause an error and stop delivering messages.

Signed-off-by: Derek Collison <derek@nats.io>

Resolves #2488 

/cc @nats-io/core


Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 4de4d0dc..4abc0c3c 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1854,6 +1854,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 			mb.dmap[seq] = struct{}{}
 			// Check if <50% utilization and minimum size met.
 			if mb.rbytes > compactMinimum && mb.rbytes>>1 > mb.bytes {
+				// FIXME(dlc) - Might want this out of band.
 				mb.compact()
 			}
 		}
@@ -1938,6 +1939,9 @@ func (mb *msgBlock) compact() {
 		return false
 	}
 
+	// For skip msgs.
+	var smh [msgHdrSize]byte
+
 	for index, lbuf := uint32(0), uint32(len(buf)); index < lbuf; {
 		if index+msgHdrSize >= lbuf {
 			return
@@ -1960,9 +1964,20 @@ func (mb *msgBlock) compact() {
 				firstSet = true
 				mb.first.seq = seq
 			}
+		} else if firstSet {
+			// This is an interior delete that we need to make sure we have a placeholder for.
+			le.PutUint32(smh[0:], emptyRecordLen)
+			le.PutUint64(smh[4:], seq|ebit)
+			le.PutUint64(smh[12:], 0)
+			le.PutUint16(smh[20:], 0)
+			nbuf = append(nbuf, smh[:]...)
+			mb.hh.Reset()
+			mb.hh.Write(smh[4:20])
+			checksum := mb.hh.Sum(nil)
+			nbuf = append(nbuf, checksum...)
 		}
 		// Always set last.
-		mb.last.seq = seq
+		mb.last.seq = seq &^ ebit
 		// Advance to next record.
 		index += rl
 	}
@@ -1986,7 +2001,9 @@ func (mb *msgBlock) compact() {
 	if err := ioutil.WriteFile(mfn, nbuf, defaultFilePerms); err != nil {
 		return
 	}
-	os.Rename(mfn, mb.mfn)
+	if err := os.Rename(mfn, mb.mfn); err != nil {
+		return
+	}
 
 	// Close cache and open FDs and index file.
 	mb.clearCacheAndOffset()
@@ -2747,6 +2764,7 @@ func (fs *fileStore) selectMsgBlock(seq uint64) *msgBlock {
 			return mb
 		}
 	}
+
 	return nil
 }
 
@@ -3168,8 +3186,10 @@ func (mb *msgBlock) cacheLookup(seq uint64) (*fileStoredMsg, error) {
 		return nil, err
 	}
 	if seq != mseq {
+		mb.cache.buf = nil
 		return nil, fmt.Errorf("sequence numbers for cache load did not match, %d vs %d", seq, mseq)
 	}
+
 	sm := &fileStoredMsg{
 		subj: subj,
 		hdr:  hdr,
@@ -3179,7 +3199,6 @@ func (mb *msgBlock) cacheLookup(seq uint64) (*fileStoredMsg, error) {
 		mb:   mb,
 		off:  int64(bi),
 	}
-
 	return sm, nil
 }
 


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 2ea1812c..5895ab16 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3266,9 +3266,6 @@ func TestFileStoreSparseCompaction(t *testing.T) {
 		if ta >= tb {
 			t.Fatalf("Expected total after to be less then before, got %d vs %d", tb, ta)
 		}
-		if ta != ua {
-			t.Fatalf("Expected compact to make total and used same, got %d vs %d", ta, ua)
-		}
 	}
 
 	// Actual testing here.
@@ -3314,3 +3311,47 @@ func TestFileStoreSparseCompaction(t *testing.T) {
 	eraseMsgs(500, 502, 504, 506, 508, 510)
 	compact()
 }
+
+func TestFileStoreSparseCompactionWithInteriorDeletes(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	cfg := StreamConfig{Name: "KV", Subjects: []string{"kv.>"}, Storage: FileStorage}
+	var fs *fileStore
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, cfg)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	for i := 1; i <= 1000; i++ {
+		if _, _, err := fs.StoreMsg(fmt.Sprintf("kv.%d", i%10), nil, []byte("OK")); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+
+	// Now do interior deletes.
+	for _, seq := range []uint64{500, 600, 700, 800} {
+		removed, err := fs.RemoveMsg(seq)
+		if err != nil || !removed {
+			t.Fatalf("Got an error on remove of %d: %v", seq, err)
+		}
+	}
+
+	_, _, _, _, err = fs.LoadMsg(900)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Do compact by hand, make sure we can still access msgs past the interior deletes.
+	fs.mu.RLock()
+	lmb := fs.lmb
+	lmb.dirtyCloseWithRemove(false)
+	lmb.compact()
+	fs.mu.RUnlock()
+
+	_, _, _, _, err = fs.LoadMsg(900)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+}