[full-ci] chore: bump reva to v2.43.0 (#2630)

vendor/github.com/nats-io/nats.go/README.md

@@ -23,7 +23,7 @@ A [Go](http://golang.org) client for the [NATS messaging system](https://nats.io
 go get github.com/nats-io/nats.go@latest
 
 # To get a specific version:
-go get github.com/nats-io/nats.go@v1.50.0
+go get github.com/nats-io/nats.go@v1.51.0
 
 # Note that the latest major version for NATS Server is v2:
 go get github.com/nats-io/nats-server/v2@latest

vendor/github.com/nats-io/nats.go/jetstream/pull.go

@@ -194,13 +194,6 @@ const (
 	unset                    = -1
 )
 
-func min(x, y int) int {
-	if x < y {
-		return x
-	}
-	return y
-}
-
 // Consume can be used to continuously receive messages and handle them
 // with the provided callback function. Consume cannot be used concurrently
 // when using ordered consumer.
@@ -258,14 +251,6 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 			}
 			return
 		}
-		defer func() {
-			sub.Lock()
-			sub.checkPending()
-			if sub.hbMonitor != nil {
-				sub.hbMonitor.Reset(2 * consumeOpts.Heartbeat)
-			}
-			sub.Unlock()
-		}()
 		if !userMsg {
 			// heartbeat message
 			if msgErr == nil {
@@ -273,15 +258,24 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 			}
 
 			sub.Lock()
-			err := sub.handleStatusMsg(msg, msgErr)
+			termErr, notifyErr := sub.handleStatusMsg(msg, msgErr)
+			if termErr == nil {
+				sub.checkPending()
+				if sub.hbMonitor != nil {
+					sub.hbMonitor.Reset(2 * consumeOpts.Heartbeat)
+				}
+			}
 			sub.Unlock()
 
-			if err != nil {
+			if sub.consumeOpts.ErrHandler != nil && notifyErr != nil {
+				sub.consumeOpts.ErrHandler(sub, notifyErr)
+			}
+			if termErr != nil {
 				if sub.closed.Load() == 1 {
 					return
 				}
 				if sub.consumeOpts.ErrHandler != nil {
-					sub.consumeOpts.ErrHandler(sub, err)
+					sub.consumeOpts.ErrHandler(sub, termErr)
 				}
 				sub.Stop()
 			}
@@ -294,6 +288,10 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 		sub.Lock()
 		sub.decrementPendingMsgs(msg)
 		sub.incrementDeliveredMsgs()
+		sub.checkPending()
+		if sub.hbMonitor != nil {
+			sub.hbMonitor.Reset(2 * consumeOpts.Heartbeat)
+		}
 		sub.Unlock()
 
 		if sub.consumeOpts.StopAfter > 0 && sub.consumeOpts.StopAfter == sub.delivered {
@@ -388,9 +386,6 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 				}
 			case err := <-sub.errs:
 				sub.Lock()
-				if sub.consumeOpts.ErrHandler != nil {
-					sub.consumeOpts.ErrHandler(sub, err)
-				}
 				if errors.Is(err, ErrNoHeartbeat) {
 					batchSize := sub.consumeOpts.MaxMessages
 					if sub.consumeOpts.StopAfter > 0 {
@@ -413,6 +408,9 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 					sub.resetPendingMsgs()
 				}
 				sub.Unlock()
+				if sub.consumeOpts.ErrHandler != nil {
+					sub.consumeOpts.ErrHandler(sub, err)
+				}
 				if errors.Is(err, ErrConnectionClosed) {
 					sub.Stop()
 				}
@@ -664,9 +662,9 @@ func (s *pullSubscription) Next(opts ...NextOpt) (Msg, error) {
 				if msgErr == nil {
 					continue
 				}
-				if err := s.handleStatusMsg(msg, msgErr); err != nil {
+				if termErr, _ := s.handleStatusMsg(msg, msgErr); termErr != nil {
 					s.Stop()
-					return nil, err
+					return nil, termErr
 				}
 				continue
 			}
@@ -715,28 +713,29 @@ func (s *pullSubscription) Next(opts ...NextOpt) (Msg, error) {
 	}
 }
 
-func (s *pullSubscription) handleStatusMsg(msg *nats.Msg, msgErr error) error {
+// handleStatusMsg processes a status message from the server.
+// It returns a terminal error (caller should stop) and a non-terminal
+// error to notify the user about via ErrHandler. The caller should invoke
+// ErrHandler outside the lock to avoid deadlocks.
+func (s *pullSubscription) handleStatusMsg(msg *nats.Msg, msgErr error) (error, error) {
 	if !errors.Is(msgErr, nats.ErrTimeout) && !errors.Is(msgErr, ErrMaxBytesExceeded) && !errors.Is(msgErr, ErrBatchCompleted) {
 		if errors.Is(msgErr, ErrConsumerDeleted) || errors.Is(msgErr, ErrBadRequest) {
-			return msgErr
+			return msgErr, nil
 		}
 		if errors.Is(msgErr, ErrPinIDMismatch) {
 			s.consumer.setPinID("")
 			s.pending.msgCount = 0
 			s.pending.byteCount = 0
 		}
-		if s.consumeOpts.ErrHandler != nil {
-			s.consumeOpts.ErrHandler(s, msgErr)
-		}
 		if errors.Is(msgErr, ErrConsumerLeadershipChanged) {
 			s.pending.msgCount = 0
 			s.pending.byteCount = 0
 		}
-		return nil
+		return nil, msgErr
 	}
 	msgsLeft, bytesLeft, err := parsePending(msg)
 	if err != nil {
-		return err
+		return err, nil
 	}
 	s.pending.msgCount -= msgsLeft
 	if s.pending.msgCount < 0 {
@@ -748,7 +747,7 @@ func (s *pullSubscription) handleStatusMsg(msg *nats.Msg, msgErr error) error {
 			s.pending.byteCount = 0
 		}
 	}
-	return nil
+	return nil, nil
 }
 
 func (hb *hbMonitor) Stop() {

vendor/github.com/nats-io/nats.go/js.go

@@ -2029,10 +2029,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 	// If maxap is greater than the default sub's pending limit, use that.
 	if maxap > DefaultSubPendingMsgsLimit {
 		// For bytes limit, use the min of maxp*1MB or DefaultSubPendingBytesLimit
-		bl := maxap * 1024 * 1024
-		if bl < DefaultSubPendingBytesLimit {
-			bl = DefaultSubPendingBytesLimit
-		}
+		bl := max(maxap*1024*1024, DefaultSubPendingBytesLimit)
 		if err := sub.SetPendingLimits(maxap, bl); err != nil {
 			return nil, err
 		}
@@ -3114,10 +3111,7 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 			}
 
 			// Make our request expiration a bit shorter than the current timeout.
-			expiresDiff := time.Duration(float64(ttl) * 0.1)
-			if expiresDiff > 5*time.Second {
-				expiresDiff = 5 * time.Second
-			}
+			expiresDiff := min(time.Duration(float64(ttl)*0.1), 5*time.Second)
 			expires := ttl - expiresDiff
 
 			nr.Batch = batch - len(msgs)
@@ -3398,10 +3392,7 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 	ttl = time.Until(deadline)
 
 	// Make our request expiration a bit shorter than the current timeout.
-	expiresDiff := time.Duration(float64(ttl) * 0.1)
-	if expiresDiff > 5*time.Second {
-		expiresDiff = 5 * time.Second
-	}
+	expiresDiff := min(time.Duration(float64(ttl)*0.1), 5*time.Second)
 	expires := ttl - expiresDiff
 
 	connStatusChanged := nc.StatusChanged()

vendor/github.com/nats-io/nats.go/nats.go

@@ -49,7 +49,7 @@ import (
 
 // Default Constants
 const (
-	Version                   = "1.50.0"
+	Version                   = "1.51.0"
 	DefaultURL                = "nats://127.0.0.1:4222"
 	DefaultPort               = 4222
 	DefaultMaxReconnect       = 60
@@ -61,6 +61,7 @@ const (
 	DefaultMaxPingOut         = 2
 	DefaultMaxChanLen         = 64 * 1024       // 64k
 	DefaultReconnectBufSize   = 8 * 1024 * 1024 // 8MB
+	DefaultWriteBufSize       = defaultBufSize
 	RequestChanLen            = 8
 	DefaultDrainTimeout       = 30 * time.Second
 	DefaultFlusherTimeout     = time.Minute
@@ -409,6 +410,34 @@ type Options struct {
 	// Defaults to 1m.
 	FlusherTimeout time.Duration
 
+	// ReconnectOnFlusherError, when set to true, causes the client to
+	// trigger a reconnect if the background flusher fails to write to the
+	// underlying connection for any reason (timeout, broken pipe,
+	// connection reset, EOF etc.).
+	//
+	// This is an advanced option. Most applications do not need to enable
+	// it: the server-side stale connection detection (via PingInterval /
+	// MaxPingsOut) and the read loop's own error handling will eventually
+	// notice a dead connection and the client will reconnect. Enable this
+	// only if you need faster recovery from a stalled or broken TCP write
+	// — for example, in latency-sensitive setups where waiting for a ping
+	// timeout is unacceptable.
+	//
+	// Messages buffered at the time of the error are lost, as they are
+	// with any flusher write error. The purpose of this option is to
+	// limit the blast radius by preventing further messages from being
+	// buffered into a potentially corrupted connection, not to recover
+	// the in-flight data.
+	//
+	// When triggered, the standard DisconnectErrHandler and
+	// ReconnectHandler callbacks are invoked as with any other reconnect.
+	// The first reconnect attempt bypasses the configured ReconnectWait
+	// so that recovery is as fast as possible; if that attempt fails,
+	// subsequent attempts obey the normal backoff.
+	//
+	// Defaults to false.
+	ReconnectOnFlusherError bool
+
 	// PingInterval is the period at which the client will be sending ping
 	// commands to the server, disabled if 0 or negative.
 	// Defaults to 2m.
@@ -574,6 +603,14 @@ type Options struct {
 	// IgnoreDiscoveredServers will disable adding advertised server URLs
 	// from INFO messages to the server pool.
 	IgnoreDiscoveredServers bool
+
+	// WriteBufferSize is an advanced option that sets the flush threshold
+	// of the write buffer used to batch outgoing data before writing to
+	// the underlying connection. In most cases, the default value should
+	// not be changed. A smaller buffer reduces the amount of data that
+	// can be lost on blocked writes but may significantly reduce throughput.
+	// Defaults to 32768 bytes (32KB).
+	WriteBufferSize int
 }
 
 const (
@@ -888,6 +925,14 @@ type ServerInfo struct {
 	Cluster      string   `json:"cluster,omitempty"`
 	ConnectURLs  []string `json:"connect_urls,omitempty"`
 	LameDuckMode bool     `json:"ldm,omitempty"`
+	// JetStream indicates whether the server has JetStream enabled.
+	JetStream bool `json:"jetstream,omitempty"`
+	// IsSystemAccount indicates whether the connected client's account
+	// is the system account.
+	IsSystemAccount bool `json:"acc_is_sys,omitempty"`
+	// JSApiLevel is the JetStream API level advertised by the server.
+	// Requires nats-server v2.12.0 or later; older servers will report 0.
+	JSApiLevel int `json:"api_lvl,omitempty"`
 }
 
 const (
@@ -1172,6 +1217,19 @@ func ReconnectBufSize(size int) Option {
 	}
 }
 
+// WriteBufferSize is an advanced option that sets the flush threshold
+// of the write buffer used to batch outgoing data before writing to
+// the underlying connection. In most cases, the default value should
+// not be changed. A smaller buffer reduces the amount of data that
+// can be lost on blocked writes but may significantly reduce throughput.
+// Defaults to 32768 bytes (32KB).
+func WriteBufferSize(size int) Option {
+	return func(o *Options) error {
+		o.WriteBufferSize = size
+		return nil
+	}
+}
+
 // Timeout is an Option to set the timeout for Dial on a connection.
 // Defaults to 2s.
 func Timeout(t time.Duration) Option {
@@ -1189,6 +1247,17 @@ func FlusherTimeout(t time.Duration) Option {
 	}
 }
 
+// ReconnectOnFlusherError is an Option to automatically trigger a
+// reconnect when the background flusher hits any write error. See
+// [Options.ReconnectOnFlusherError] for details. This is an
+// advanced option and is usually not required.
+func ReconnectOnFlusherError() Option {
+	return func(o *Options) error {
+		o.ReconnectOnFlusherError = true
+		return nil
+	}
+}
+
 // DrainTimeout is an Option to set the timeout for draining a connection.
 // Defaults to 30s.
 func DrainTimeout(t time.Duration) Option {
@@ -1741,6 +1810,10 @@ func (o Options) Connect() (*Conn, error) {
 	if nc.Opts.ReconnectBufSize == 0 {
 		nc.Opts.ReconnectBufSize = DefaultReconnectBufSize
 	}
+	// Default WriteBufferSize
+	if nc.Opts.WriteBufferSize <= 0 {
+		nc.Opts.WriteBufferSize = DefaultWriteBufSize
+	}
 	// Ensure that Timeout is not 0
 	if nc.Opts.Timeout == 0 {
 		nc.Opts.Timeout = DefaultTimeout
@@ -2080,7 +2153,7 @@ func (nc *Conn) newReaderWriter() {
 		off: -1,
 	}
 	nc.bw = &natsWriter{
-		limit:  defaultBufSize,
+		limit:  nc.Opts.WriteBufferSize,
 		plimit: nc.Opts.ReconnectBufSize,
 	}
 }
@@ -2414,8 +2487,10 @@ func (nc *Conn) ForceReconnect() error {
 	// Stop ping timer if set.
 	nc.stopPingTimer()
 
-	// Go ahead and make sure we have flushed the outbound
+	// flush any pending data and switch to pending mode to buffer new outgoing
+	// data until we reconnect and can flush it.
 	nc.bw.flush()
+	nc.bw.switchToPending()
 	nc.conn.Close()
 
 	nc.changeConnStatus(RECONNECTING)
@@ -2574,6 +2649,40 @@ func (nc *Conn) ConnectedClusterName() string {
 	return nc.info.Cluster
 }
 
+// ConnectedServerJetStream reports whether the connected server has
+// JetStream enabled and, if so, its API level. The API level is
+// advertised by nats-server v2.12.0 or later; older servers will
+// report 0 even when JetStream is enabled.
+func (nc *Conn) ConnectedServerJetStream() (bool, int) {
+	if nc == nil {
+		return false, 0
+	}
+
+	nc.mu.RLock()
+	defer nc.mu.RUnlock()
+
+	if nc.status != CONNECTED {
+		return false, 0
+	}
+	return nc.info.JetStream, nc.info.JSApiLevel
+}
+
+// IsSystemAccount reports whether the connected client's account
+// is the system account.
+func (nc *Conn) IsSystemAccount() bool {
+	if nc == nil {
+		return false
+	}
+
+	nc.mu.RLock()
+	defer nc.mu.RUnlock()
+
+	if nc.status != CONNECTED {
+		return false
+	}
+	return nc.info.IsSystemAccount
+}
+
 // Low level setup for structs, etc
 func (nc *Conn) setup() {
 	nc.subs = make(map[int64]*Subscription)
@@ -3316,8 +3425,10 @@ func (nc *Conn) doReconnect(err error, forceReconnect bool) {
 }
 
 // processOpErr handles errors from reading or parsing the protocol.
-// The lock should not be held entering this function.
-func (nc *Conn) processOpErr(err error) bool {
+// The lock should not be held entering this function. If forceReconnect
+// is true, the first reconnect attempt will bypass the configured
+// ReconnectWait; subsequent attempts still obey the normal backoff.
+func (nc *Conn) processOpErr(err error, forceReconnect bool) bool {
 	nc.mu.Lock()
 	defer nc.mu.Unlock()
 	if nc.isConnecting() || nc.isClosed() || nc.isReconnecting() {
@@ -3340,7 +3451,7 @@ func (nc *Conn) processOpErr(err error) bool {
 		// Clear any queued pongs, e.g. pending flush calls.
 		nc.clearPendingFlushCalls()
 
-		go nc.doReconnect(err, false)
+		go nc.doReconnect(err, forceReconnect)
 		return false
 	}
 
@@ -3443,7 +3554,7 @@ func (nc *Conn) readLoop() {
 			err = nc.parse(buf)
 		}
 		if err != nil {
-			if shouldClose := nc.processOpErr(err); shouldClose {
+			if shouldClose := nc.processOpErr(err, false); shouldClose {
 				nc.close(CLOSED, true, nil)
 			}
 			break
@@ -3891,6 +4002,13 @@ func (nc *Conn) flusher() {
 				if asyncErrorCB := nc.Opts.AsyncErrorCB; asyncErrorCB != nil {
 					nc.ach.push(func() { asyncErrorCB(nc, nil, err) })
 				}
+				if nc.Opts.ReconnectOnFlusherError {
+					nc.mu.Unlock()
+					if shouldClose := nc.processOpErr(err, true); shouldClose {
+						nc.close(CLOSED, true, nil)
+					}
+					return
+				}
 			}
 		}
 		nc.mu.Unlock()
@@ -4070,11 +4188,11 @@ func (nc *Conn) processErr(ie string) {
 
 	// FIXME(dlc) - process Slow Consumer signals special.
 	if e == STALE_CONNECTION {
-		close = nc.processOpErr(ErrStaleConnection)
+		close = nc.processOpErr(ErrStaleConnection, false)
 	} else if e == MAX_CONNECTIONS_ERR {
-		close = nc.processOpErr(ErrMaxConnectionsExceeded)
+		close = nc.processOpErr(ErrMaxConnectionsExceeded, false)
 	} else if e == MAX_ACCOUNT_CONNECTIONS_ERR {
-		close = nc.processOpErr(ErrMaxAccountConnectionsExceeded)
+		close = nc.processOpErr(ErrMaxAccountConnectionsExceeded, false)
 	} else if strings.HasPrefix(e, PERMISSIONS_ERR) {
 		nc.processTransientError(fmt.Errorf("%w: %s", ErrPermissionViolation, ne))
 	} else if strings.HasPrefix(e, MAX_SUBSCRIPTIONS_ERR) {
@@ -5656,7 +5774,7 @@ func (nc *Conn) processPingTimer() {
 	nc.pout++
 	if nc.pout > nc.Opts.MaxPingsOut {
 		nc.mu.Unlock()
-		if shouldClose := nc.processOpErr(ErrStaleConnection); shouldClose {
+		if shouldClose := nc.processOpErr(ErrStaleConnection, false); shouldClose {
 			nc.close(CLOSED, true, nil)
 		}
 		return

vendor/github.com/nats-io/nats.go/ws.go

@@ -55,6 +55,12 @@ const (
 	wsMaxControlPayloadSize = 125
 	wsCloseSatusSize        = 2
 
+	// wsMaxMsgPayloadMultiple is the multiplier applied to MaxPayload to
+	// determine the maximum WebSocket frame size.
+	wsMaxMsgPayloadMultiple = 8
+	// wsMaxMsgPayloadLimit is the absolute cap on WebSocket frame size (64MB).
+	wsMaxMsgPayloadLimit = 64 * 1024 * 1024
+
 	// From https://tools.ietf.org/html/rfc6455#section-11.7
 	wsCloseStatusNormalClosure      = 1000
 	wsCloseStatusNoStatusReceived   = 1005
@@ -114,10 +120,7 @@ func (d *wsDecompressor) Read(dst []byte) (int, error) {
 	copied := 0
 	rem := len(dst)
 	for buf := d.bufs[0]; buf != nil && rem > 0; {
-		n := len(buf[d.off:])
-		if n > rem {
-			n = rem
-		}
+		n := min(len(buf[d.off:]), rem)
 		copy(dst[copied:], buf[d.off:d.off+n])
 		copied += n
 		rem -= n
@@ -182,6 +185,18 @@ func wsNewReader(r io.Reader) *websocketReader {
 	return &websocketReader{r: r, ff: true}
 }
 
+// maxFrameSize returns the maximum allowed WebSocket frame size based on the
+// negotiated MaxPayload. This mirrors the server-side wsMaxMessageSize logic.
+func (r *websocketReader) maxFrameSize() uint64 {
+	if r.nc != nil {
+		mp := r.nc.info.MaxPayload
+		if mp > 0 && uint64(mp) <= wsMaxMsgPayloadLimit/wsMaxMsgPayloadMultiple {
+			return uint64(mp) * wsMaxMsgPayloadMultiple
+		}
+	}
+	return wsMaxMsgPayloadLimit
+}
+
 // From now on, reads will be from the readLoop and we will need to
 // acquire the connection lock should we have to send/write a control
 // message from handleControlFrame.
@@ -288,7 +303,14 @@ func (r *websocketReader) Read(p []byte) (int, error) {
 			if err != nil {
 				return 0, err
 			}
-			rem = int(binary.BigEndian.Uint64(tmpBuf))
+			rem64 := binary.BigEndian.Uint64(tmpBuf)
+			if rem64&(1<<63) != 0 {
+				return 0, errors.New("invalid websocket frame: MSB set in 64-bit payload length")
+			}
+			if rem64 > r.maxFrameSize() {
+				return 0, fmt.Errorf("websocket frame too large: %d", rem64)
+			}
+			rem = int(rem64)
 		}
 
 		// Handle control messages in place...