Repository: nats-io/nats-server

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]))
```

Discussion:
@ripienaar: Server 2.9.20 is now quite a while out of date, let us know how latest 2.10 works for you.
@nenych: Below you can see the same test with the NATS 2.10.14, with this version we have even worse results:

![CleanShot 2024-05-07 at 18 36 04@2x](https://github.com/nats-io/nats-server/assets/10981686/fe5167fd-4dd7-4804-b3d7-c3a589e098c3)
@kam1kaze: Any updates here? We have the same issue on our cluster. Thanks@nenych: Hello, we still are observing this issue, when at least 1 slow consumer is detected we have up to 90% performance degradation.
On the screenshot below we had 1 slow pod.
Server version: 2.10.19-RC.3-alpine3.20
![CleanShot 2024-09-30 at 11 07 55@2x](https://github.com/user-attachments/assets/2636c004-ad95-4a21-9690-21d9b416de9f)
@derekcollison: @nenych are you a Synadia customer?@nenych: @derekcollison  No, I am not.@derekcollison: No worries, we will always do our best to help out the ecosystem. We do prioritize customers of course.

We would need to do a video call with you I think as a next step to really understand what is going on.@nenych: @derekcollison Sure, we can have a video call. Right now we have some test infrastructure where can show you the problem and our findings.@derekcollison: Will see if @wallyqs has some time to jump on a call.@wallyqs: Hi @nenych ping me at `wally@nats.io` when you are available and can have a look.@kozlovic: I think it is just due to the detection of consumer(s) that are falling behind and the server stalls the fast producers. Running the server in Debug mode (-D) should show you messages similar to `Timed out of fast producer stall (100ms)`. It affects the inbound of messages from producers on that subject (meaning the subject matching the slow consumer(s)). But it would not affect other producers that send to non slow consumers (well aside that there is a maximum that the server can handle so the perf per producer may decrease but overall inbound perf increase or be maintained).

That has always been the case (although we did tweak the stalling approach along the years).
@alexbozhenko: I tested on 2.10.26 using an awesome reproducer provided by @nenych, and seems like #6568 did not make any difference in this test:

![Image](https://github.com/user-attachments/assets/4a1bbb3b-462f-44a1-baf7-df88176af69d)
Yellow lines mark start and stop of a slow consumer@wallyqs: yes there is still quite a bit of impact but it does make a difference compared to before, in this setup using `no_fast_producer_stall` would produce better results@alexbozhenko: Thanks, @wallyqs.
I confirm that with `no_fast_producer_stall` option(added in https://github.com/nats-io/nats-server/pull/6500, documented in https://github.com/nats-io/nats.docs/pull/811) slow consumer doesn't affect the rate of messaged passing through the system 

(Here I tested with CPU and memory limits in the container). 
When a consumer switched to be slow, the message rate is not affected:
![Image](https://github.com/user-attachments/assets/6584f39b-21df-4a19-9f6d-a5737f36a142)
https://github.com/alexbozhenko/nats-test/tree/signalled_consumer



Solution 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)
+}