Drop JetStream API requests when request queue built up

Signed-off-by: Neil Twigg <[email protected]>

server/jetstream_api.go

@@ -824,11 +824,9 @@ func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, sub
 
 	// Copy the state. Note the JSAPI only uses the hdr index to piece apart the
 	// header from the msg body. No other references are needed.
-	// Check pending and warn if getting backed up.
-	const warnThresh = 128
-	pending, _ := s.jsAPIRoutedReqs.push(&jsAPIRoutedReq{jsub, sub, acc, subject, reply, copyBytes(rmsg), c.pa})
-	if pending >= warnThresh {
-		s.rateLimitFormatWarnf("JetStream request queue has high pending count: %d", pending)
+	_, err := s.jsAPIRoutedReqs.push(&jsAPIRoutedReq{jsub, sub, acc, subject, reply, copyBytes(rmsg), c.pa})
+	if err == errIPQLenLimitReached {
+		s.RateLimitWarnf("Dropping JetStream API requests: %s", err)
 	}
 }
 
@@ -871,12 +869,16 @@ func (s *Server) setJetStreamExportSubs() error {
 	// Start the go routine that will process API requests received by the
 	// subscription below when they are coming from routes, etc..
 	const maxProcs = 16
+	const maxRequests = 10_000
 	mp := runtime.GOMAXPROCS(0)
 	// Cap at 16 max for now on larger core setups.
 	if mp > maxProcs {
 		mp = maxProcs
 	}
-	s.jsAPIRoutedReqs = newIPQueue[*jsAPIRoutedReq](s, "Routed JS API Requests")
+	s.jsAPIRoutedReqs = newIPQueue[*jsAPIRoutedReq](
+		s, "Routed JS API Requests",
+		ipqLimitByLen[*jsAPIRoutedReq](maxRequests),
+	)
 	for i := 0; i < mp; i++ {
 		s.startGoRoutine(s.processJSAPIRoutedRequests)
 	}