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

Issue:
Performance degradation with slow consumers [v2.9.X, v2.10.X]
### Observed behavior

Performance degradation after the slow consumer connection. 
As you can see below, we are observing about 30% degradation of the incoming messages when the slow consumer connected, and about 50% after the second one.

![CleanShot 2024-05-07 at 16 58 44@2x](https://github.com/nats-io/nats-server/assets/10981686/d91b6175-6934-431f-851b-b74f13e09149)


### Expected behavior

Stop sending messages to the slow consumers until their buffers are empty without slowing down the server.

### Server and client version

Server: 2.9.20
Python library: nats-py 2.7.2

### Host environment

**Local:**
MacOS 14.4.1, arm64, Docker 26.0.0
The same behavior with the amd64 emulator (--platform=linux/amd64 flag).

**GKE**
Container-Optimized OS, amd64, containerd


### Steps to reproduce

Prepared required configs and docker-compose file that will start NATS, Prometheus, an exporter, and two consumers: [https://github.com/nenych/nats-test](https://github.com/nenych/nats-test).

#### Steps to run
1. Clone the repository.
2. Build the docker image:
```
docker build -t test/nats:latest .
```
3. Install NATS cli: https://docs.nats.io/using-nats/nats-tools/nats_cli
4. Run docker-compose (will start NATS, prometheus and 2 consumers):
```
docker-compose -f ./docker-compose.yaml up -d
```
5. Start NATS benchmark:
```
nats bench updates --pub=4 --msgs 1000000000 --size=1000
```
6. Wait a little and start the slow consumer:
```
docker run --rm -it --network=nats-test_default test/nats:latest python3 slow-consumer.py
```

#### Explore metrics
1. Open prometheus: http://localhost:9091/graph
2. Insert query:
```
sum by (job) (rate(nats_varz_in_msgs[30s]))
```

Pull request description:
[IMPROVED] Adjusted stalled producers to not be as penalizing.
Fixes #5394

Signed-off-by: Derek Collison <derek@nats.io>


Patch:
diff --git a/server/client.go b/server/client.go
index 62f4b714..16e24491 100644
--- a/server/client.go
+++ b/server/client.go
@@ -113,8 +113,9 @@ const (
 	maxNoRTTPingBeforeFirstPong = 2 * time.Second
 
 	// For stalling fast producers
-	stallClientMinDuration = 100 * time.Millisecond
-	stallClientMaxDuration = time.Second
+	stallClientMinDuration = 2 * time.Millisecond
+	stallClientMaxDuration = 5 * time.Millisecond
+	stallTotalAllowed      = 10 * time.Millisecond
 )
 
 var readLoopReportThreshold = readLoopReport
@@ -464,6 +465,9 @@ type readCache struct {
 
 	// Capture the time we started processing our readLoop.
 	start time.Time
+
+	// Total time stalled so far for readLoop processing.
+	tst time.Duration
 }
 
 // set the flag (would be equivalent to set the boolean to true)
@@ -1416,6 +1420,11 @@ func (c *client) readLoop(pre []byte) {
 				}
 				return
 			}
+			// Clear total stalled time here.
+			if c.in.tst >= stallClientMaxDuration {
+				c.rateLimitFormatWarnf("Producer was stalled for a total of %v", c.in.tst.Round(time.Millisecond))
+			}
+			c.in.tst = 0
 		}
 
 		// If we are a ROUTER/LEAF and have processed an INFO, it is possible that
@@ -1734,7 +1743,7 @@ func (c *client) flushOutbound() bool {
 
 	// Check if we have a stalled gate and if so and we are recovering release
 	// any stalled producers. Only kind==CLIENT will stall.
-	if c.out.stc != nil && (n == attempted || c.out.pb < c.out.mp/2) {
+	if c.out.stc != nil && (n == attempted || c.out.pb < c.out.mp/4*3) {
 		close(c.out.stc)
 		c.out.stc = nil
 	}
@@ -2327,7 +2336,8 @@ func (c *client) queueOutbound(data []byte) {
 	// Check here if we should create a stall channel if we are falling behind.
 	// We do this here since if we wait for consumer's writeLoop it could be
 	// too late with large number of fan in producers.
-	if c.out.pb > c.out.mp/2 && c.out.stc == nil {
+	// If the outbound connection is > 75% of maximum pending allowed, create a stall gate.
+	if c.out.pb > c.out.mp/4*3 && c.out.stc == nil {
 		c.out.stc = make(chan struct{})
 	}
 }
@@ -3382,31 +3392,36 @@ func (c *client) msgHeader(subj, reply []byte, sub *subscription) []byte {
 }
 
 func (c *client) stalledWait(producer *client) {
+	// Check to see if we have exceeded our total wait time per readLoop invocation.
+	if producer.in.tst > stallTotalAllowed {
+		return
+	}
+
+	// Grab stall channel which the slow consumer will close when caught up.
 	stall := c.out.stc
-	ttl := stallDuration(c.out.pb, c.out.mp)
+
 	c.mu.Unlock()
 	defer c.mu.Lock()
 
+	// Calculate stall time.
+	ttl := stallClientMinDuration
+	if c.out.pb >= c.out.mp {
+		ttl = stallClientMaxDuration
+	}
+	// Now check if we are close to total allowed.
+	if producer.in.tst+ttl > stallTotalAllowed {
+		ttl = stallTotalAllowed - producer.in.tst
+	}
 	delay := time.NewTimer(ttl)
 	defer delay.Stop()
 
+	start := time.Now()
 	select {
 	case <-stall:
 	case <-delay.C:
 		producer.Debugf("Timed out of fast producer stall (%v)", ttl)
 	}
-}
-
-func stallDuration(pb, mp int64) time.Duration {
-	ttl := stallClientMinDuration
-	if pb >= mp {
-		ttl = stallClientMaxDuration
-	} else if hmp := mp / 2; pb > hmp {
-		bsz := hmp / 10
-		additional := int64(ttl) * ((pb - hmp) / bsz)
-		ttl += time.Duration(additional)
-	}
-	return ttl
+	producer.in.tst += time.Since(start)
 }
 
 // Used to treat maps as efficient set


Test patch:
diff --git a/server/client_test.go b/server/client_test.go
index 8a8d7a4f..3f8821f0 100644
--- a/server/client_test.go
+++ b/server/client_test.go
@@ -2061,32 +2061,6 @@ func TestClientNoSlowConsumerIfConnectExpected(t *testing.T) {
 	}
 }
 
-func TestClientStalledDuration(t *testing.T) {
-	for _, test := range []struct {
-		name        string
-		pb          int64
-		mp          int64
-		expectedTTL time.Duration
-	}{
-		{"pb above mp", 110, 100, stallClientMaxDuration},
-		{"pb equal mp", 100, 100, stallClientMaxDuration},
-		{"pb below mp/2", 49, 100, stallClientMinDuration},
-		{"pb equal mp/2", 50, 100, stallClientMinDuration},
-		{"pb at 55% of mp", 55, 100, stallClientMinDuration + 1*stallClientMinDuration},
-		{"pb at 60% of mp", 60, 100, stallClientMinDuration + 2*stallClientMinDuration},
-		{"pb at 70% of mp", 70, 100, stallClientMinDuration + 4*stallClientMinDuration},
-		{"pb at 80% of mp", 80, 100, stallClientMinDuration + 6*stallClientMinDuration},
-		{"pb at 90% of mp", 90, 100, stallClientMinDuration + 8*stallClientMinDuration},
-		{"pb at 99% of mp", 99, 100, stallClientMinDuration + 9*stallClientMinDuration},
-	} {
-		t.Run(test.name, func(t *testing.T) {
-			if ttl := stallDuration(test.pb, test.mp); ttl != test.expectedTTL {
-				t.Fatalf("For pb=%v mp=%v, expected TTL to be %v, got %v", test.pb, test.mp, test.expectedTTL, ttl)
-			}
-		})
-	}
-}
-
 func TestClientIPv6Address(t *testing.T) {
 	opts := DefaultOptions()
 	opts.Host = "0.0.0.0"
diff --git a/server/msgtrace_test.go b/server/msgtrace_test.go
index ddf78bab..7e61549a 100644
--- a/server/msgtrace_test.go
+++ b/server/msgtrace_test.go
@@ -1,4 +1,4 @@
-// Copyright 2024 The NATS Authors
+// Copyright 2024-2025 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
@@ -564,10 +564,8 @@ func TestMsgTraceEgressErrors(t *testing.T) {
 				}
 				msg.Data = []byte("hello")
 				nc2.PublishMsg(msg)
-				time.Sleep(10 * time.Millisecond)
-				cid, err = nc2.GetClientID()
-				require_NoError(t, err)
-				c = s.GetClient(cid)
+				// This needs to be less than default stall time, which now is 2ms.
+				time.Sleep(1 * time.Millisecond)
 				c.mu.Lock()
 				c.flags.set(closeConnection)
 				c.mu.Unlock()
diff --git a/server/norace_test.go b/server/norace_test.go
index 44d0d677..6e558413 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -11476,3 +11476,61 @@ func TestNoRaceNoFastProducerStall(t *testing.T) {
 	// Should have been delivered to the trace subscription.
 	checkTraceMsg(_EMPTY_)
 }
+
+func TestNoRaceProducerStallLimits(t *testing.T) {
+	tmpl := `
+		listen: "127.0.0.1:-1"
+	`
+	conf := createConfFile(t, []byte(tmpl))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	ncSlow := natsConnect(t, s.ClientURL())
+	defer ncSlow.Close()
+	natsSub(t, ncSlow, "foo", func(m *nats.Msg) { m.Respond([]byte("42")) })
+	natsFlush(t, ncSlow)
+
+	ncProd := natsConnect(t, s.ClientURL())
+	defer ncProd.Close()
+
+	cid, err := ncSlow.GetClientID()
+	require_NoError(t, err)
+	c := s.GetClient(cid)
+	require_True(t, c != nil)
+
+	// Artificially set a stall channel on the subscriber.
+	c.mu.Lock()
+	c.out.stc = make(chan struct{})
+	c.mu.Unlock()
+
+	start := time.Now()
+	_, err = ncProd.Request("foo", []byte("HELLO"), time.Second)
+	elapsed := time.Since(start)
+	require_NoError(t, err)
+
+	// This should have not cleared on its own but should have bettwen min and max pause.
+	require_True(t, elapsed >= stallClientMinDuration)
+	require_True(t, elapsed < stallClientMaxDuration)
+
+	// Now test total maximum by loading up a bunch of requests and measuring the last one.
+	// Artificially set a stall channel again on the subscriber.
+	c.mu.Lock()
+	c.out.stc = make(chan struct{})
+	// This will prevent us from clearing the stc.
+	c.out.pb = c.out.mp/4*3 + 100
+	c.mu.Unlock()
+
+	for i := 0; i < 10; i++ {
+		err = ncProd.PublishRequest("foo", "bar", []byte("HELLO"))
+		require_NoError(t, err)
+	}
+	start = time.Now()
+	_, err = ncProd.Request("foo", []byte("HELLO"), time.Second)
+	elapsed = time.Since(start)
+	require_NoError(t, err)
+
+	require_True(t, elapsed >= stallTotalAllowed)
+	// Should always be close to totalAllowed (e.g. 10ms), but if you run alot of them in one go can bump up
+	// just past 12ms, hence the Max setting below to avoid a flapper.
+	require_True(t, elapsed < stallTotalAllowed+stallClientMaxDuration)
+}