Repository: nats-io/nats-server
Pull request #2183 (2021-05-05 14:51:53)

[FIXED] Stream byte limit setting failed on dynamic account, filestore not honoring blksize.
 Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/errors.go b/server/errors.go
index 15eff79c..684cce51 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -1,4 +1,4 @@
-// Copyright 2012-2020 The NATS Authors
+// Copyright 2012-2021 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -202,6 +202,12 @@ var (
 
 	// ErrJetStreamResourcesExceeded is returned when a call would exceed internal resource limits.
 	ErrJetStreamResourcesExceeded = errors.New("jetstream resources exceeded for server")
+
+	// ErrStorageResourcesExceeded is returned when storage resources would be exceeded.
+	ErrStorageResourcesExceeded = errors.New("insufficient storage resources available")
+
+	// ErrMemoryResourcesExceeded is returned when memory resources would be exceeded.
+	ErrMemoryResourcesExceeded = errors.New("insufficient memory resources available")
 )
 
 // configErr is a configuration error.
diff --git a/server/filestore.go b/server/filestore.go
index e20d816c..3bada57a 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -97,8 +97,9 @@ type msgBlock struct {
 	ifn     string
 	ifd     *os.File
 	liwsz   int64
-	index   uint64
-	bytes   uint64
+	index   uint64 // User visible message count.
+	bytes   uint64 // User visible bytes count.
+	rbytes  uint64 // Total bytes (raw) including deleted. Used for rolling to new blk.
 	msgs    uint64
 	lwits   int64
 	lwts    int64
@@ -410,6 +411,12 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint64) *msgBlock {
 	}
 	defer file.Close()
 
+	if fi, _ := file.Stat(); fi != nil {
+		mb.rbytes = uint64(fi.Size())
+	} else {
+		return nil
+	}
+
 	// Read our index file. Use this as source of truth if possible.
 	if err := mb.readIndexInfo(); err == nil {
 		// Quick sanity check here.
@@ -479,7 +486,7 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 	startLastSeq := mb.last.seq
 
 	// Clear state we need to rebuild.
-	mb.msgs, mb.bytes = 0, 0
+	mb.msgs, mb.bytes, mb.rbytes = 0, 0, 0
 	mb.last.seq, mb.last.ts = 0, 0
 	firstNeedsSet := true
 
@@ -487,6 +494,7 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 	if err != nil {
 		return nil, err
 	}
+	mb.rbytes = uint64(len(buf))
 
 	addToDmap := func(seq uint64) {
 		if seq == 0 {
@@ -614,6 +622,7 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 
 			mb.msgs++
 			mb.bytes += uint64(rl)
+			mb.rbytes += uint64(rl)
 		}
 
 		index += rl
@@ -1579,6 +1588,8 @@ func (mb *msgBlock) truncate(sm *fileStoredMsg) (nmsgs, nbytes uint64, err error
 	// Do local mb stat updates.
 	mb.msgs -= purged
 	mb.bytes -= bytes
+	mb.rbytes -= bytes
+
 	// Update our last msg.
 	mb.last.seq = sm.seq
 	mb.last.ts = sm.ts
@@ -2002,10 +2013,10 @@ func (mb *msgBlock) bytesPending() ([]byte, error) {
 	return buf, nil
 }
 
-// Return the number of bytes in this message block.
-func (mb *msgBlock) numBytes() uint64 {
+// Returns the current blkSize including deleted msgs etc.
+func (mb *msgBlock) blkSize() uint64 {
 	mb.mu.RLock()
-	nb := mb.bytes
+	nb := mb.rbytes
 	mb.mu.RUnlock()
 	return nb
 }
@@ -2021,6 +2032,7 @@ func (mb *msgBlock) updateAccounting(seq uint64, ts int64, rl uint64) {
 	atomic.StoreUint64(&mb.last.seq, seq)
 	mb.last.ts = ts
 	mb.bytes += rl
+	mb.rbytes += rl
 	mb.msgs++
 }
 
@@ -2035,7 +2047,7 @@ func (fs *fileStore) writeMsgRecord(seq uint64, ts int64, subj string, hdr, msg
 	}
 	// Grab our current last message block.
 	mb := fs.lmb
-	if mb == nil || mb.numBytes()+rl > fs.fcfg.BlockSize {
+	if mb == nil || mb.blkSize()+rl > fs.fcfg.BlockSize {
 		if mb, err = fs.newMsgBlockForWrite(); err != nil {
 			return 0, err
 		}
diff --git a/server/jetstream.go b/server/jetstream.go
index b0dcdff2..db25b1b3 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1317,14 +1317,33 @@ func (jsa *jsAccount) checkLimits(config *StreamConfig) error {
 func (jsa *jsAccount) checkBytesLimits(addBytes int64, storage StorageType) error {
 	switch storage {
 	case MemoryStorage:
-		if jsa.memReserved+addBytes > jsa.limits.MaxMemory {
-			return fmt.Errorf("insufficient memory resources available")
+		// Account limits defined.
+		if jsa.limits.MaxMemory > 0 {
+			if jsa.memReserved+addBytes > jsa.limits.MaxMemory {
+				return ErrMemoryResourcesExceeded
+			}
+		} else {
+			// Account is unlimited, check if this server can handle request.
+			js := jsa.js
+			if js.memReserved+addBytes > js.config.MaxMemory {
+				return ErrMemoryResourcesExceeded
+			}
 		}
 	case FileStorage:
-		if jsa.storeReserved+addBytes > jsa.limits.MaxStore {
-			return fmt.Errorf("insufficient storage resources available")
+		// Account limits defined.
+		if jsa.limits.MaxStore > 0 {
+			if jsa.storeReserved+addBytes > jsa.limits.MaxStore {
+				return ErrStorageResourcesExceeded
+			}
+		} else {
+			// Account is unlimited, check if this server can handle request.
+			js := jsa.js
+			if js.storeReserved+addBytes > js.config.MaxStore {
+				return ErrStorageResourcesExceeded
+			}
 		}
 	}
+
 	return nil
 }
 
@@ -1421,11 +1440,12 @@ func (js *jetStream) sufficientResources(limits *JetStreamAccountLimits) error {
 	if limits == nil {
 		return nil
 	}
+
 	if js.memReserved+limits.MaxMemory > js.config.MaxMemory {
-		return fmt.Errorf("insufficient memory resources available")
+		return ErrMemoryResourcesExceeded
 	}
 	if js.storeReserved+limits.MaxStore > js.config.MaxStore {
-		return fmt.Errorf("insufficient storage resources available")
+		return ErrStorageResourcesExceeded
 	}
 	return nil
 }


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index d4e41324..8dabc318 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -2833,3 +2833,52 @@ func TestFileStoreStreamPurgeAndDirtyRestartBug(t *testing.T) {
 		t.Fatalf("Unexpected state: %+v", state)
 	}
 }
+
+// rip
+func TestFileStoreStreamFailToRollBug(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 512},
+		StreamConfig{Name: "zzz", Storage: FileStorage, MaxBytes: 300},
+	)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	// Make sure we properly roll underlying blocks.
+	n, msg := 200, bytes.Repeat([]byte("ABC"), 33) // ~100bytes
+	for i := 0; i < n; i++ {
+		if _, _, err := fs.StoreMsg("zzz", nil, msg); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+
+	// Grab some info for introspection.
+	fs.mu.RLock()
+	numBlks := len(fs.blks)
+	var index uint64
+	var blkSize int64
+	if numBlks > 0 {
+		mb := fs.blks[0]
+		mb.mu.RLock()
+		index = mb.index
+		if fi, _ := os.Stat(mb.mfn); fi != nil {
+			blkSize = fi.Size()
+		}
+		mb.mu.RUnlock()
+	}
+	fs.mu.RUnlock()
+
+	if numBlks != 1 {
+		t.Fatalf("Expected only one block, got %d", numBlks)
+	}
+	if index < 60 {
+		t.Fatalf("Expected a block index > 60, got %d", index)
+	}
+	if blkSize > 512 {
+		t.Fatalf("Expected block to be <= 512, got %d", blkSize)
+	}
+}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 280daf41..5158acff 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -114,6 +114,36 @@ func TestJetStreamClusterAccountInfo(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterStreamLimitWithAccountDefaults(t *testing.T) {
+	// 2MB memory, 8MB disk
+	c := createJetStreamClusterWithTemplate(t, jsClusterLimitsTempl, "R3L", 3)
+	defer c.shutdown()
+
+	// Client based API
+	s := c.randomNonLeader()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo", "bar"},
+		Replicas: 2,
+		MaxBytes: 4 * 1024 * 1024,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST2",
+		Replicas: 2,
+		MaxBytes: 15 * 1024 * 1024,
+	})
+	if err == nil || !strings.Contains(err.Error(), "insufficient storage") {
+		t.Fatalf("Expected %v but got %v", ErrStorageResourcesExceeded, err)
+	}
+}
+
 func TestJetStreamClusterSingleReplicaStreams(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R1S", 3)
 	defer c.shutdown()