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

Issue:
Making the producer stalling configurable
### Proposed change

Making the producer stalling configurable.

Currently, when you have a slow consumer on a subject, it can make the server throttle the provider.
ex: 
```
// If we are a client and we detect that the consumer we are
// sending to is in a stalled state, go ahead and wait here
// with a limit.
if c.kind == CLIENT && client.out.stc != nil {
client.stalledWait(c)
}
```

I read on another issue that it was to protect GW and Routes ?
The proposal would be to make this configurable by allowing people to switch this off if they need.

### Use case

Better handling event spikes by not stalling the producers if there are a slow consumer on some topic.

### Contribution

The change looks not huge here but we would love to have some more inputs before saying we can do it.

Pull request:
[ADDED] Option that disables fast producer stalling (drops msg instead)
Normally, when a producer detects that one of the consumer of a message is falling behind, it will stall. Which means that if a message has 2 consumers and the first is "slow", then it will affect the timely delivery to the second consumer.

With the new option `no_fast_producer_stall=true`, the server will simply drop a message destined to a consumer that would have caused the producer to stall. The message is still delivered to consumers that are not falling behind.

The option can be config-reload'ed and if a message is dropped due to fast-producer/slow-consumer, and the message was traced (with deliver option), then the message trace egress event will have an error indicating the reason why the message was not delivered.

Resolves #6413

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>


Patch:
diff --git a/server/client.go b/server/client.go
index 73df4a49..4f872d01 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3586,6 +3586,11 @@ func (c *client) deliverMsg(prodIsMQTT bool, sub *subscription, acc *Account, su
 	// sending to is in a stalled state, go ahead and wait here
 	// with a limit.
 	if c.kind == CLIENT && client.out.stc != nil {
+		if srv.getOpts().NoFastProducerStall {
+			mt.addEgressEvent(client, sub, errMsgTraceFastProdNoStall)
+			client.mu.Unlock()
+			return false
+		}
 		client.stalledWait(c)
 	}
 
diff --git a/server/msgtrace.go b/server/msgtrace.go
index b7e229ba..e3e73421 100644
--- a/server/msgtrace.go
+++ b/server/msgtrace.go
@@ -244,6 +244,7 @@ const (
 	errMsgTraceSubClosed       = "Not delivered because subscription is closed"
 	errMsgTraceClientClosed    = "Not delivered because client is closed"
 	errMsgTraceAutoSubExceeded = "Not delivered because auto-unsubscribe exceeded"
+	errMsgTraceFastProdNoStall = "Not delivered because fast producer not stalled and consumer is slow"
 )
 
 type msgTrace struct {
diff --git a/server/opts.go b/server/opts.go
index 19c26dc1..3ea672a3 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -320,6 +320,7 @@ type Options struct {
 	MaxControlLine             int32         `json:"max_control_line"`
 	MaxPayload                 int32         `json:"max_payload"`
 	MaxPending                 int64         `json:"max_pending"`
+	NoFastProducerStall        bool          `json:"-"`
 	Cluster                    ClusterOpts   `json:"cluster,omitempty"`
 	Gateway                    GatewayOpts   `json:"gateway,omitempty"`
 	LeafNode                   LeafNodeOpts  `json:"leaf,omitempty"`
@@ -1674,6 +1675,8 @@ func (o *Options) processConfigFileLine(k string, v any, errors *[]error, warnin
 			*errors = append(*errors, err)
 			return
 		}
+	case "no_fast_producer_stall":
+		o.NoFastProducerStall = v.(bool)
 	default:
 		if au := atomic.LoadInt32(&allowUnknownTopLevelField); au == 0 && !tk.IsUsedVariable() {
 			err := &unknownConfigFieldErr{
diff --git a/server/reload.go b/server/reload.go
index 9c8b8b56..c5740ebd 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -917,6 +917,19 @@ func (l *leafNodeOption) Apply(s *Server) {
 	}
 }
 
+type noFastProdStallReload struct {
+	noopOption
+	noStall bool
+}
+
+func (l *noFastProdStallReload) Apply(s *Server) {
+	var not string
+	if l.noStall {
+		not = "not "
+	}
+	s.Noticef("Reloaded: fast producers will %sbe stalled", not)
+}
+
 // Compares options and disconnects clients that are no longer listed in pinned certs. Lock must not be held.
 func (s *Server) recheckPinnedCerts(curOpts *Options, newOpts *Options) {
 	s.mu.Lock()
@@ -1638,6 +1651,8 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			// skip changes in config digest, this is handled already while
 			// processing the config.
 			continue
+		case "nofastproducerstall":
+			diffOpts = append(diffOpts, &noFastProdStallReload{noStall: newValue.(bool)})
 		default:
 			// TODO(ik): Implement String() on those options to have a nice print.
 			// %v is difficult to figure what's what, %+v print private fields and


Test patch:
diff --git a/server/norace_test.go b/server/norace_test.go
index 78f0da1c..c488f772 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -11365,3 +11365,217 @@ func TestNoRaceStoreReverseWalkWithDeletesPerf(t *testing.T) {
 		}
 	}
 }
+
+type fastProdLogger struct {
+	DummyLogger
+	gotIt chan struct{}
+}
+
+func (l *fastProdLogger) Debugf(format string, args ...any) {
+	msg := fmt.Sprintf(format, args...)
+	if strings.Contains(msg, "fast producer") {
+		select {
+		case l.gotIt <- struct{}{}:
+		default:
+		}
+	}
+}
+
+type slowWriteConn struct {
+	sync.RWMutex
+	net.Conn
+	delay bool
+}
+
+func (c *slowWriteConn) Write(b []byte) (int, error) {
+	c.RLock()
+	delay := c.delay
+	c.RUnlock()
+	if delay {
+		time.Sleep(100 * time.Millisecond)
+	}
+	return c.Conn.Write(b)
+}
+
+func TestNoRaceNoFastProducerStall(t *testing.T) {
+	tmpl := `
+		listen: "127.0.0.1:-1"
+		no_fast_producer_stall: %s
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, "true")))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	l := &fastProdLogger{gotIt: make(chan struct{}, 1)}
+	s.SetLogger(l, true, false)
+
+	ncSlow := natsConnect(t, s.ClientURL(), nats.ReconnectWait(10*time.Millisecond))
+	defer ncSlow.Close()
+	natsSub(t, ncSlow, "foo", func(_ *nats.Msg) {})
+	natsFlush(t, ncSlow)
+
+	cid, err := ncSlow.GetClientID()
+	require_NoError(t, err)
+	c := s.GetClient(cid)
+	require_True(t, c != nil)
+
+	c.mu.Lock()
+	swc := &slowWriteConn{Conn: c.nc, delay: true}
+	c.nc = swc
+	c.mu.Unlock()
+
+	defer func() {
+		swc.Lock()
+		swc.delay = false
+		swc.Unlock()
+	}()
+
+	// The producer could still overwhelm the "fast" consumer.
+	// So we will send more than what it will use as a target
+	// to consider the test done.
+	total := 2_000_000
+	target := total / 2
+	ch := make(chan struct{}, 1)
+	var count int
+
+	ncFast := natsConnect(t, s.ClientURL(), nats.ReconnectWait(10*time.Millisecond))
+	defer ncFast.Close()
+	fastSub := natsSub(t, ncFast, "foo", func(_ *nats.Msg) {
+		if count++; count == target {
+			ch <- struct{}{}
+		}
+	})
+	err = fastSub.SetPendingLimits(-1, -1)
+	require_NoError(t, err)
+	natsFlush(t, ncFast)
+
+	ncProd := natsConnect(t, s.ClientURL(), nats.ReconnectWait(10*time.Millisecond))
+	defer ncProd.Close()
+
+	cid, err = ncProd.GetClientID()
+	require_NoError(t, err)
+	pc := s.GetClient(cid)
+	pc.mu.Lock()
+	pcnc := pc.nc
+	pc.mu.Unlock()
+
+	payload := make([]byte, 256)
+
+	for i := 0; i < total; i++ {
+		natsPub(t, ncProd, "foo", payload)
+	}
+	select {
+	case <-ch:
+		// OK
+	case <-time.After(10 * time.Second):
+		t.Fatal("Test timed-out")
+	}
+	// Now wait a bit and make sure we did not get any fast producer debug statements.
+	select {
+	case <-l.gotIt:
+		t.Fatal("Got debug logs about fast producer")
+	case <-time.After(time.Second):
+		// OK
+	}
+
+	// We don't need that one anymore
+	ncFast.Close()
+	// Now we will conf reload to enable fast producer stalling.
+	reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, "false"))
+
+	// Since the producer can block, we will publish from a different routine,
+	// and check for the debug trace from the main.
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	doneCh := make(chan struct{})
+	go func() {
+		defer wg.Done()
+		for i := 0; ; i++ {
+			ncProd.Publish("foo", payload)
+			select {
+			case <-doneCh:
+				return
+			default:
+			}
+			if i%1000 == 0 {
+				time.Sleep(time.Millisecond)
+			}
+		}
+	}()
+	select {
+	case <-l.gotIt:
+		pcnc.Close()
+		close(doneCh)
+	case <-time.After(20 * time.Second):
+		t.Fatal("Timed-out waiting for a warning")
+	}
+	wg.Wait()
+}
+
+func TestNoRaceNoFastProducerStallAndMsgTrace(t *testing.T) {
+	o := DefaultOptions()
+	o.NoFastProducerStall = true
+	o.WriteDeadline = 2 * time.Second
+	s := RunServer(o)
+	defer s.Shutdown()
+
+	ncSlow := natsConnect(t, s.ClientURL(), nats.ReconnectWait(10*time.Millisecond))
+	defer ncSlow.Close()
+	natsSub(t, ncSlow, "foo", func(_ *nats.Msg) {})
+	natsFlush(t, ncSlow)
+
+	cid, err := ncSlow.GetClientID()
+	require_NoError(t, err)
+	c := s.GetClient(cid)
+	require_True(t, c != nil)
+
+	c.mu.Lock()
+	swc := &slowWriteConn{Conn: c.nc, delay: true}
+	c.nc = swc
+	c.mu.Unlock()
+
+	defer func() {
+		swc.Lock()
+		swc.delay = false
+		swc.Unlock()
+	}()
+
+	ncProd := natsConnect(t, s.ClientURL(), nats.ReconnectWait(10*time.Millisecond))
+	defer ncProd.Close()
+
+	payload := make([]byte, 256)
+
+	nc := natsConnect(t, s.ClientURL(), nats.ReconnectWait(10*time.Millisecond))
+	defer nc.Close()
+	doneCh := make(chan struct{})
+	traceSub := natsSub(t, nc, "my.trace.subj", func(traceMsg *nats.Msg) {
+		var e MsgTraceEvent
+		json.Unmarshal(traceMsg.Data, &e)
+		egresses := e.Egresses()
+		if len(egresses) == 1 {
+			if err := egresses[0].Error; err == errMsgTraceFastProdNoStall {
+				close(doneCh)
+				traceMsg.Sub.Unsubscribe()
+			}
+		}
+	})
+	err = traceSub.SetPendingLimits(-1, -1)
+	require_NoError(t, err)
+	natsFlush(t, nc)
+
+	msg := nats.NewMsg("foo")
+	msg.Header.Set(MsgTraceDest, traceSub.Subject)
+	msg.Data = payload
+	for i, done := 0, false; !done; i++ {
+		ncProd.PublishMsg(msg)
+		select {
+		case <-doneCh:
+			// OK
+			return
+		default:
+		}
+		if i%1000 == 0 {
+			time.Sleep(time.Millisecond)
+		}
+	}
+}