Repository: nats-io/nats-server
Pull request #3418 (2022-08-30 19:56:07)

[ADDED] JetStream: Ability to configure the per server max catchup bytes
The original value was hardcoded to 128MB and 32MB per stream. The
per-server limit is lowered to 32MB but is configurable with
a new configuration parameter:
```
jetstream {
   max_catchup: 8MB
}
```

The per-stream limit was also lowered from 32MB/128,000msgs to
8MB/32,000 messages.

Tests have shown no difference in performance for fast links.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/server/jetstream.go b/server/jetstream.go
index 0eb03aec..95add36f 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -322,6 +322,11 @@ func (s *Server) checkStoreDir(cfg *JetStreamConfig) error {
 // enableJetStream will start up the JetStream subsystem.
 func (s *Server) enableJetStream(cfg JetStreamConfig) error {
 	js := &jetStream{srv: s, config: cfg, accounts: make(map[string]*jsAccount), apiSubs: NewSublistNoCache()}
+	s.gcbMu.Lock()
+	if s.gcbOutMax = s.getOpts().JetStreamMaxCatchup; s.gcbOutMax == 0 {
+		s.gcbOutMax = defaultMaxTotalCatchupOutBytes
+	}
+	s.gcbMu.Unlock()
 
 	s.mu.Lock()
 	s.js = js
@@ -2674,6 +2679,10 @@ func validateJetStreamOptions(o *Options) error {
 	default:
 		return fmt.Errorf("expected 'no_extend' for string value, got '%s'", h)
 	}
+
+	if o.JetStreamMaxCatchup < 0 {
+		return fmt.Errorf("jetstream max catchup cannot be negative")
+	}
 	return nil
 }
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index dfa7976f..135adedd 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -7177,7 +7177,7 @@ func (mset *stream) handleClusterStreamInfoRequest(sub *subscription, c *client,
 	sysc.sendInternalMsg(reply, _EMPTY_, nil, si)
 }
 
-const maxTotalCatchupOutBytes = int64(128 * 1024 * 1024) // 128MB for now, for the total server.
+const defaultMaxTotalCatchupOutBytes = int64(32 * 1024 * 1024) // 32MB for now, for the total server.
 
 // Current total outstanding catchup bytes.
 func (s *Server) gcbTotal() int64 {
@@ -7186,6 +7186,14 @@ func (s *Server) gcbTotal() int64 {
 	return s.gcbOut
 }
 
+// Returns true if Current total outstanding catchup bytes is below
+// the maximum configured.
+func (s *Server) gcbBelowMax() bool {
+	s.gcbMu.RLock()
+	defer s.gcbMu.RUnlock()
+	return s.gcbOut <= s.gcbOutMax
+}
+
 // Adds `sz` to the server's total outstanding catchup bytes and to `localsz`
 // under the gcbMu lock. The `localsz` points to the local outstanding catchup
 // bytes of the runCatchup go routine of a given stream.
@@ -7193,7 +7201,7 @@ func (s *Server) gcbAdd(localsz *int64, sz int64) {
 	s.gcbMu.Lock()
 	atomic.AddInt64(localsz, sz)
 	s.gcbOut += sz
-	if s.gcbOut >= maxTotalCatchupOutBytes && s.gcbKick == nil {
+	if s.gcbOut >= s.gcbOutMax && s.gcbKick == nil {
 		s.gcbKick = make(chan struct{})
 	}
 	s.gcbMu.Unlock()
@@ -7209,7 +7217,7 @@ func (s *Server) gcbSubLocked(localsz *int64, sz int64) {
 	}
 	atomic.AddInt64(localsz, -sz)
 	s.gcbOut -= sz
-	if s.gcbKick != nil && s.gcbOut < maxTotalCatchupOutBytes {
+	if s.gcbKick != nil && s.gcbOut < s.gcbOutMax {
 		close(s.gcbKick)
 		s.gcbKick = nil
 	}
@@ -7244,8 +7252,8 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 	s := mset.srv
 	defer s.grWG.Done()
 
-	const maxOutBytes = int64(32 * 1024 * 1024) // 32MB for now, these are all internal, from server to server
-	const maxOutMsgs = int32(128 * 1024)
+	const maxOutBytes = int64(8 * 1024 * 1024) // 8MB for now, these are all internal, from server to server
+	const maxOutMsgs = int32(32 * 1024)
 	outb := int64(0)
 	outm := int32(0)
 
@@ -7346,7 +7354,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 
 		var smv StoreMsg
 
-		for ; seq <= last && atomic.LoadInt64(&outb) <= maxOutBytes && atomic.LoadInt32(&outm) <= maxOutMsgs && s.gcbTotal() <= maxTotalCatchupOutBytes; seq++ {
+		for ; seq <= last && atomic.LoadInt64(&outb) <= maxOutBytes && atomic.LoadInt32(&outm) <= maxOutMsgs && s.gcbBelowMax(); seq++ {
 			sm, err := mset.store.LoadMsg(seq, &smv)
 			// if this is not a deleted msg, bail out.
 			if err != nil && err != ErrStoreMsgNotFound && err != errDeletedMsg {
diff --git a/server/opts.go b/server/opts.go
index ecfa1d5e..d2cf0fc3 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -255,6 +255,7 @@ type Options struct {
 	JetStreamCipher       StoreCipher   `json:"-"`
 	JetStreamUniqueTag    string
 	JetStreamLimits       JSLimitOpts
+	JetStreamMaxCatchup   int64
 	StoreDir              string            `json:"-"`
 	JsAccDefaultDomain    map[string]string `json:"-"` // account to domain name mapping
 	Websocket             WebsocketOpts     `json:"-"`
@@ -1948,6 +1949,12 @@ func parseJetStream(v interface{}, opts *Options, errors *[]error, warnings *[]e
 				}
 			case "unique_tag":
 				opts.JetStreamUniqueTag = strings.ToLower(strings.TrimSpace(mv.(string)))
+			case "max_outstanding_catchup":
+				s, err := getStorageSize(mv)
+				if err != nil {
+					return &configErr{tk, fmt.Sprintf("%s %s", strings.ToLower(mk), err)}
+				}
+				opts.JetStreamMaxCatchup = s
 			default:
 				if !tk.IsUsedVariable() {
 					err := &unknownConfigFieldErr{
diff --git a/server/server.go b/server/server.go
index d4b677c6..bd9762db 100644
--- a/server/server.go
+++ b/server/server.go
@@ -279,8 +279,9 @@ type Server struct {
 	rateLimitLoggingCh chan time.Duration
 
 	// Total outstanding catchup bytes in flight.
-	gcbMu  sync.RWMutex
-	gcbOut int64
+	gcbMu     sync.RWMutex
+	gcbOut    int64
+	gcbOutMax int64 // Taken from JetStreamMaxCatchup or defaultMaxTotalCatchupOutBytes
 	// A global chanel to kick out stalled catchup sequences.
 	gcbKick chan struct{}
 


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index c873f8be..2fbfa212 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -13066,3 +13066,92 @@ func TestJetStreamClusterReplicasChangeStreamInfo(t *testing.T) {
 
 	checkStreamInfo(js)
 }
+
+func TestJetStreamClusterMaxOutstandingCatchup(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "MCB", 3)
+	defer c.shutdown()
+
+	for _, s := range c.servers {
+		s.gcbMu.RLock()
+		v := s.gcbOutMax
+		s.gcbMu.RUnlock()
+		if v != defaultMaxTotalCatchupOutBytes {
+			t.Fatalf("Server %v, expected max_outstanding_catchup to be %v, got %v", s, defaultMaxTotalCatchupOutBytes, v)
+		}
+	}
+
+	c.shutdown()
+
+	tmpl := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: { max_outstanding_catchup: 1KB, domain: ngs, max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+		leaf: { listen: 127.0.0.1:-1 }
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+	`
+	c = createJetStreamClusterWithTemplate(t, tmpl, "MCB", 3)
+	defer c.shutdown()
+
+	for _, s := range c.servers {
+		s.gcbMu.RLock()
+		v := s.gcbOutMax
+		s.gcbMu.RUnlock()
+		if v != 1024 {
+			t.Fatalf("Server %v, expected max_outstanding_catchup to be 1KB, got %v", s, v)
+		}
+	}
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+	// Close client now and will create new one
+	nc.Close()
+
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	follower := c.randomNonStreamLeader(globalAccountName, "TEST")
+	follower.Shutdown()
+
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	// Create new connection in case we would have been connected to follower.
+	nc, _ = jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	payload := string(make([]byte, 2048))
+	for i := 0; i < 1000; i++ {
+		sendStreamMsg(t, nc, "foo", payload)
+	}
+
+	// Cause snapshots on leader
+	mset, err := c.streamLeader(globalAccountName, "TEST").GlobalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	err = mset.raftNode().InstallSnapshot(mset.stateSnapshot())
+	require_NoError(t, err)
+
+	// Resart server and it should be able to catchup
+	follower = c.restartServer(follower)
+	c.waitOnStreamCurrent(follower, globalAccountName, "TEST")
+
+	// Config reload not supported
+	s := c.servers[0]
+	cfile := s.getOpts().ConfigFile
+	content, err := os.ReadFile(cfile)
+	require_NoError(t, err)
+	conf := string(content)
+	conf = strings.ReplaceAll(conf, "max_outstanding_catchup: 1KB,", "max_outstanding_catchup: 1MB,")
+	err = os.WriteFile(cfile, []byte(conf), 0644)
+	require_NoError(t, err)
+	err = s.Reload()
+	require_Error(t, err, fmt.Errorf("config reload not supported for JetStreamMaxCatchup: old=1024, new=1048576"))
+}