Repository: nats-io/nats-server
Pull request #1695 (2020-11-08 23:27:54)

Clarify HPUB and HMSG examples
 adding more examples


Patch:
diff --git a/GOVERNANCE.md b/GOVERNANCE.md
index 186d3d4f..152e14af 100644
--- a/GOVERNANCE.md
+++ b/GOVERNANCE.md
@@ -1,3 +1,3 @@
 # NATS Server Governance
 
-NATS Server (gnastd) is part of the NATS project and is subject to the [NATS Governance](https://github.com/nats-io/nats-general/blob/master/GOVERNANCE.md).
\ No newline at end of file
+NATS Server is part of the NATS project and is subject to the [NATS Governance](https://github.com/nats-io/nats-general/blob/master/GOVERNANCE.md).
diff --git a/doc/adr/0004-nats-headers.md b/doc/adr/0004-nats-headers.md
index ba8d5e9b..134fbc75 100644
--- a/doc/adr/0004-nats-headers.md
+++ b/doc/adr/0004-nats-headers.md
@@ -57,16 +57,16 @@ Messages that include a header have a `HPUB` protocol:
 ```
 HPUB SUBJECT REPLY 23 30␍␊NATS/1.0␍␊Header: X␍␊␍␊PAYLOAD␍␊
 HPUB SUBJECT REPLY 23 23␍␊NATS/1.0␍␊Header: X␍␊␍␊␍␊
+HPUB SUBJECT REPLY 48 55␍␊NATS/1.0␍␊Header1: X␍␊Header1: Y␍␊Header2: Z␍␊␍␊PAYLOAD␍␊
+HPUB SUBJECT REPLY 48 48␍␊NATS/1.0␍␊Header1: X␍␊Header1: Y␍␊Header2: Z␍␊␍␊␍␊
 
 HPUB <SUBJ> [REPLY] <HDR_LEN> <TOT_LEN>
 <HEADER><PAYLOAD>
-
 ```
 
 #### NOTES:
-
-- `HDR_LEN` includes the entire serialized header
-- `TOT_LEN` includes the entire payload length including the header
+- `HDR_LEN` includes the entire serialized header, from the start of the version string up to and including the ␍␊ before the payload
+- `TOT_LEN` the payload length plus the HDR_LEN
 
 ### MSG with Headers
 
@@ -74,15 +74,16 @@ Clients will see `HMSG` protocol lines for `MSG`s that contain headers
 
 ```
 HMSG SUBJECT 1 REPLY 23 30␍␊NATS/1.0␍␊Header: X␍␊␍␊PAYLOAD␍␊
-HPUB SUBJECT 1 REPLY 23 23␍␊NATS/1.0␍␊Header: X␍␊␍␊␍␊
+HMSG SUBJECT 1 REPLY 23 23␍␊NATS/1.0␍␊Header: X␍␊␍␊␍␊
+HMSG SUBJECT 1 REPLY 48 55␍␊NATS/1.0␍␊Header1: X␍␊Header1: Y␍␊Header2: Z␍␊␍␊PAYLOAD␍␊
+HMSG SUBJECT 1 REPLY 48 48␍␊NATS/1.0␍␊Header1: X␍␊Header1: Y␍␊Header2: Z␍␊␍␊␍␊
 
 HMSG <SUBJECT> <SID> [REPLY] <HDR_LEN> <TOT_LEN>
 <PAYLOAD>
 ```
 
-- `HDR_LEN` includes the entire header
-- `TOT_LEN` includes the entire payload length including the header
-
+- `HDR_LEN` includes the entire serialized header, from the start of the version string up to and including the ␍␊ before the payload
+- `TOT_LEN` the payload length plus the HDR_LEN
 
 ## Decision
 
diff --git a/server/const.go b/server/const.go
index c5ef63d0..50f694d0 100644
--- a/server/const.go
+++ b/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.30"
+	VERSION = "2.2.0-beta.31"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/filestore.go b/server/filestore.go
index 6dcf9e07..e0645f1f 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -826,12 +826,11 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 
 	mb.mu.Lock()
 
-	// Check cache. This will be very rare, will hold lock on this one.
+	// Check cache. This will be very rare.
 	if mb.cache == nil || mb.cache.idx == nil {
 		mb.mu.Unlock()
 		fs.mu.Unlock()
 		if err := mb.loadMsgs(); err != nil {
-			fs.mu.Unlock()
 			return false, err
 		}
 		fs.mu.Lock()
@@ -940,7 +939,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 // Grab info from a slot.
 // Lock should be held.
 func (mb *msgBlock) slotInfo(slot int) (uint32, uint32, bool, error) {
-	if mb.cache == nil || mb.cache.idx == nil {
+	if mb.cache == nil || slot >= len(mb.cache.idx) {
 		return 0, 0, false, errPartialCache
 	}
 	bi := mb.cache.idx[slot]
@@ -1969,16 +1968,18 @@ func (mb *msgBlock) cacheLookupWithLock(seq uint64) (*fileStoredMsg, error) {
 		}
 	}
 
-	// Update cache activity.
-	mb.llts = time.Now().UnixNano()
-
-	bi, _, hashChecked, _ := mb.slotInfo(int(seq - mb.cache.fseq))
+	if mb.cache.off > 0 {
+		return nil, errPartialCache
+	}
 
-	// Check if partial cache and we miss.
-	if mb.cache.off > 0 && bi <= uint32(mb.cache.off) {
+	bi, _, hashChecked, err := mb.slotInfo(int(seq - mb.cache.fseq))
+	if err != nil {
 		return nil, errPartialCache
 	}
 
+	// Update cache activity.
+	mb.llts = time.Now().UnixNano()
+
 	// We use the high bit to denote we have already checked the checksum.
 	var hh hash.Hash64
 	if !hashChecked {


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 4c967770..bd5885ec 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -1358,6 +1358,54 @@ func TestFileStorePartialCacheExpiration(t *testing.T) {
 	}
 }
 
+func TestFileStorePartialIndexes(t *testing.T) {
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	cexp := 10 * time.Millisecond
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: cexp}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	toSend := 5
+	for i := 0; i < toSend; i++ {
+		fs.StoreMsg("foo", nil, []byte("ok-1"))
+	}
+
+	// Now wait til the cache expires, including the index.
+	fs.mu.Lock()
+	mb := fs.blks[0]
+	fs.mu.Unlock()
+
+	// Force idx to expire by resetting last remove ts.
+	mb.mu.Lock()
+	mb.lrts = mb.lrts - int64(defaultCacheIdxExpiration*2)
+	mb.mu.Unlock()
+
+	checkFor(t, time.Second, 10*time.Millisecond, func() error {
+		mb.mu.Lock()
+		defer mb.mu.Unlock()
+		if mb.cache == nil || len(mb.cache.idx) == 0 {
+			return nil
+		}
+		return fmt.Errorf("Index not empty")
+	})
+
+	// Create a partial cache by adding more msgs.
+	for i := 0; i < toSend; i++ {
+		fs.StoreMsg("foo", nil, []byte("ok-2"))
+	}
+	// If we now load in a message in second half if we do not
+	// detect idx is a partial correctly this will panic.
+	if _, _, _, _, err := fs.LoadMsg(8); err != nil {
+		t.Fatalf("Error loading %d: %v", 1, err)
+	}
+}
+
 func TestFileStoreSnapshot(t *testing.T) {
 	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
 	os.MkdirAll(storeDir, 0755)