Switch routed JS API queue to LIFO

Signed-off-by: Neil Twigg <neil@nats.io>

Author: neilalexander

server/ipqueue.go

@@ -207,6 +207,43 @@ func (q *ipQueue[T]) popOne() (T, bool) {
 	return e, true
 }
 
+// Returns the last element from the queue, if any. See comment above
+// regarding calling after being notified that there is something and
+// the use of drain(). In short, the caller should always check the
+// boolean return value to ensure that the value is genuine and not a
+// default empty value.
+func (q *ipQueue[T]) popOneLast() (T, bool) {
+	q.Lock()
+	l := len(q.elts) - q.pos
+	if l == 0 {
+		q.Unlock()
+		var empty T
+		return empty, false
+	}
+	e := q.elts[len(q.elts)-1]
+	q.elts = q.elts[:len(q.elts)-1]
+	if l--; l > 0 {
+		if q.calc != nil {
+			q.sz -= q.calc(e)
+		}
+		// We need to re-signal
+		select {
+		case q.ch <- struct{}{}:
+		default:
+		}
+	} else {
+		// We have just emptied the queue, so we can reuse unless it is too big.
+		if cap(q.elts) <= q.mrs {
+			q.elts = q.elts[:0]
+		} else {
+			q.elts = nil
+		}
+		q.pos, q.sz = 0, 0
+	}
+	q.Unlock()
+	return e, true
+}
+
 // After a pop(), the slice can be recycled for the next push() when
 // a first element is added to the queue.
 // This will also decrement the "in progress" count with the length

server/jetstream_api.go

@@ -883,14 +883,13 @@ func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, sub
 	// If we are here we have received this request over a non-client connection.
 	// We need to make sure not to block. We will send the request to a long-lived
 	// pool of go routines.
-
-	// Increment inflight. Do this before queueing.
-	atomic.AddInt64(&js.apiInflight, 1)
-
 	// 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.
-	pending, _ := s.jsAPIRoutedReqs.push(&jsAPIRoutedReq{jsub, sub, acc, subject, reply, copyBytes(rmsg), c.pa})
+	pending, err := s.jsAPIRoutedReqs.push(&jsAPIRoutedReq{jsub, sub, acc, subject, reply, copyBytes(rmsg), c.pa})
+	if err == nil {
+		atomic.AddInt64(&js.apiInflight, 1)
+	}
 	limit := atomic.LoadInt64(&js.queueLimit)
 	if pending >= int(limit) {
 		s.rateLimitFormatWarnf("JetStream API queue limit reached, dropping %d requests", pending)
@@ -926,7 +925,7 @@ func (s *Server) processJSAPIRoutedRequests() {
 			// Only pop one item at a time here, otherwise if the system is recovering
 			// from queue buildup, then one worker will pull off all the tasks and the
 			// others will be starved of work.
-			for r, ok := queue.popOne(); ok && r != nil; r, ok = queue.popOne() {
+			for r, ok := queue.popOneLast(); ok && r != nil; r, ok = queue.popOneLast() {
 				client.pa = r.pa
 				start := time.Now()
 				r.jsub.icb(r.sub, client, r.acc, r.subject, r.reply, r.msg)