bump reva and deps

Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de>
This commit is contained in:
Jörn Friedrich Dreyer
2025-02-28 17:40:07 +01:00
parent 63c4f0b1b4
commit a901ab860a
132 changed files with 3558 additions and 1479 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2020 The NATS Authors
// Copyright 2020-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2013-2018 The NATS Authors
// Copyright 2013-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2019 The NATS Authors
// Copyright 2012-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2019 The NATS Authors
// Copyright 2012-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2018 The NATS Authors
// Copyright 2012-2024 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
+9 -4
View File
@@ -1,4 +1,4 @@
// Copyright 2018-2023 The NATS Authors
// Copyright 2018-2024 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
@@ -858,9 +858,14 @@ func (a *Account) Interest(subject string) int {
func (a *Account) addClient(c *client) int {
a.mu.Lock()
n := len(a.clients)
if a.clients != nil {
a.clients[c] = struct{}{}
// Could come here earlier than the account is registered with the server.
// Make sure we can still track clients.
if a.clients == nil {
a.clients = make(map[*client]struct{})
}
a.clients[c] = struct{}{}
// If we did not add it, we are done
if n == len(a.clients) {
a.mu.Unlock()
@@ -2021,7 +2026,7 @@ func (a *Account) addServiceImportSub(si *serviceImport) error {
a.mu.Unlock()
cb := func(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
c.processServiceImport(si, acc, msg)
c.pa.delivered = c.processServiceImport(si, acc, msg)
}
sub, err := c.processSubEx([]byte(subject), nil, []byte(sid), cb, true, true, false)
if err != nil {
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2022-2023 The NATS Authors
// Copyright 2022-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 The NATS Authors
// Copyright 2023-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2022-2023 The NATS Authors
// Copyright 2022-2024 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
@@ -1,4 +1,4 @@
// Copyright 2022-2024 The NATS Authors
// Copyright 2022-2023 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2016-2018 The NATS Authors
// Copyright 2016-2020 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
+91 -40
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2023 The NATS Authors
// Copyright 2012-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
@@ -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
@@ -462,6 +463,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)
@@ -1414,6 +1418,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
@@ -1640,8 +1649,10 @@ func (c *client) flushOutbound() bool {
}
consumed := len(wnb)
// Actual write to the socket.
nc.SetWriteDeadline(start.Add(wdl))
// Actual write to the socket. The deadline applies to each batch
// rather than the total write, such that the configured deadline
// can be tuned to a known maximum quantity (64MB).
nc.SetWriteDeadline(time.Now().Add(wdl))
wn, err = wnb.WriteTo(nc)
nc.SetWriteDeadline(time.Time{})
@@ -1728,7 +1739,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
}
@@ -2290,7 +2301,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{})
}
}
@@ -3335,31 +3347,37 @@ 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)
// Calculate stall time.
ttl := stallClientMinDuration
if c.out.pb >= c.out.mp {
ttl = stallClientMaxDuration
}
c.mu.Unlock()
defer c.mu.Lock()
// 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
@@ -3451,10 +3469,15 @@ func (c *client) deliverMsg(prodIsMQTT bool, sub *subscription, acc *Account, su
msgSize -= int64(LEN_CR_LF)
}
// No atomic needed since accessed under client lock.
// Monitor is reading those also under client's lock.
client.outMsgs++
client.outBytes += msgSize
// We do not update the outbound stats if we are doing trace only since
// this message will not be sent out.
// Also do not update on internal callbacks.
if sub.icb == nil {
// No atomic needed since accessed under client lock.
// Monitor is reading those also under client's lock.
client.outMsgs++
client.outBytes += msgSize
}
// Check for internal subscriptions.
if sub.icb != nil && !c.noIcb {
@@ -3465,23 +3488,35 @@ func (c *client) deliverMsg(prodIsMQTT bool, sub *subscription, acc *Account, su
}
client.mu.Unlock()
// For service imports, track if we delivered.
didDeliver := true
// Internal account clients are for service imports and need the '\r\n'.
start := time.Now()
if client.kind == ACCOUNT {
sub.icb(sub, c, acc, string(subject), string(reply), msg)
// If we are a service import check to make sure we delivered the message somewhere.
if sub.si {
didDeliver = c.pa.delivered
}
} else {
sub.icb(sub, c, acc, string(subject), string(reply), msg[:msgSize])
}
if dur := time.Since(start); dur >= readLoopReportThreshold {
srv.Warnf("Internal subscription on %q took too long: %v", subject, dur)
}
return true
return didDeliver
}
// 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 {
if srv.getOpts().NoFastProducerStall {
client.mu.Unlock()
return false
}
client.stalledWait(c)
}
@@ -3959,7 +3994,7 @@ func (c *client) processInboundClientMsg(msg []byte) (bool, bool) {
reply = append(reply, '@')
reply = append(reply, c.pa.deliver...)
}
didDeliver = c.sendMsgToGateways(acc, msg, c.pa.subject, reply, qnames) || didDeliver
didDeliver = c.sendMsgToGateways(acc, msg, c.pa.subject, reply, qnames, false) || didDeliver
}
// Check to see if we did not deliver to anyone and the client has a reply subject set
@@ -4006,7 +4041,7 @@ func (c *client) handleGWReplyMap(msg []byte) bool {
reply = append(reply, '@')
reply = append(reply, c.pa.deliver...)
}
c.sendMsgToGateways(c.acc, msg, c.pa.subject, reply, nil)
c.sendMsgToGateways(c.acc, msg, c.pa.subject, reply, nil, false)
}
return true
}
@@ -4129,9 +4164,20 @@ func (c *client) setHeader(key, value string, msg []byte) []byte {
return bb.Bytes()
}
// Will return the value for the header denoted by key or nil if it does not exists.
// This function ignores errors and tries to achieve speed and no additional allocations.
// Will return a copy of the value for the header denoted by key or nil if it does not exist.
// If you know that it is safe to refer to the underlying hdr slice for the period that the
// return value is used, then sliceHeader() will be faster.
func getHeader(key string, hdr []byte) []byte {
v := sliceHeader(key, hdr)
if v == nil {
return nil
}
return append(make([]byte, 0, len(v)), v...)
}
// Will return the sliced value for the header denoted by key or nil if it does not exists.
// This function ignores errors and tries to achieve speed and no additional allocations.
func sliceHeader(key string, hdr []byte) []byte {
if len(hdr) == 0 {
return nil
}
@@ -4156,15 +4202,14 @@ func getHeader(key string, hdr []byte) []byte {
index++
}
// Collect together the rest of the value until we hit a CRLF.
var value []byte
start := index
for index < hdrLen {
if hdr[index] == '\r' && index < hdrLen-1 && hdr[index+1] == '\n' {
break
}
value = append(value, hdr[index])
index++
}
return value
return hdr[start:index:index]
}
// For bytes.HasPrefix below.
@@ -4175,17 +4220,17 @@ var (
// processServiceImport is an internal callback when a subscription matches an imported service
// from another account. This includes response mappings as well.
func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byte) {
func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byte) bool {
// If we are a GW and this is not a direct serviceImport ignore.
isResponse := si.isRespServiceImport()
if (c.kind == GATEWAY || c.kind == ROUTER) && !isResponse {
return
return false
}
// Detect cycles and ignore (return) when we detect one.
if len(c.pa.psi) > 0 {
for i := len(c.pa.psi) - 1; i >= 0; i-- {
if psi := c.pa.psi[i]; psi.se == si.se {
return
return false
}
}
}
@@ -4206,7 +4251,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
// response service imports and rrMap entries which all will need to simply expire.
// TODO(dlc) - Come up with something better.
if shouldReturn || (checkJS && si.se != nil && si.se.acc == c.srv.SystemAccount()) {
return
return false
}
var nrr []byte
@@ -4278,7 +4323,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
var ci *ClientInfo
if hadPrevSi && c.pa.hdr >= 0 {
var cis ClientInfo
if err := json.Unmarshal(getHeader(ClientInfoHdr, msg[:c.pa.hdr]), &cis); err == nil {
if err := json.Unmarshal(sliceHeader(ClientInfoHdr, msg[:c.pa.hdr]), &cis); err == nil {
ci = &cis
ci.Service = acc.Name
// Check if we are moving into a share details account from a non-shared
@@ -4287,7 +4332,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
c.addServerAndClusterInfo(ci)
}
}
} else if c.kind != LEAF || c.pa.hdr < 0 || len(getHeader(ClientInfoHdr, msg[:c.pa.hdr])) == 0 {
} else if c.kind != LEAF || c.pa.hdr < 0 || len(sliceHeader(ClientInfoHdr, msg[:c.pa.hdr])) == 0 {
ci = c.getClientInfo(share)
// If we did not share but the imports destination is the system account add in the server and cluster info.
if !share && isSysImport {
@@ -4345,7 +4390,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
flags |= pmrCollectQueueNames
var queues [][]byte
didDeliver, queues = c.processMsgResults(siAcc, rr, msg, c.pa.deliver, []byte(to), nrr, flags)
didDeliver = c.sendMsgToGateways(siAcc, msg, []byte(to), nrr, queues) || didDeliver
didDeliver = c.sendMsgToGateways(siAcc, msg, []byte(to), nrr, queues, false) || didDeliver
} else {
didDeliver, _ = c.processMsgResults(siAcc, rr, msg, c.pa.deliver, []byte(to), nrr, flags)
}
@@ -4354,6 +4399,10 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
c.in.rts = orts
c.pa = pacopy
// Before we undo didDeliver based on tracing and last mile, mark in the c.pa which informs us of no responders status.
// If we override due to tracing and traceOnly we do not want to send back a no responders.
c.pa.delivered = didDeliver
// Determine if we should remove this service import. This is for response service imports.
// We will remove if we did not deliver, or if we are a response service import and we are
// a singleton, or we have an EOF message.
@@ -4383,6 +4432,8 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
siAcc.removeRespServiceImport(rsi, reason)
}
}
return didDeliver
}
func (c *client) addSubToRouteTargets(sub *subscription) {
@@ -4846,7 +4897,7 @@ func (c *client) checkLeafClientInfoHeader(msg []byte) (dmsg []byte, setHdr bool
if c.pa.hdr < 0 || len(msg) < c.pa.hdr {
return msg, false
}
cir := getHeader(ClientInfoHdr, msg[:c.pa.hdr])
cir := sliceHeader(ClientInfoHdr, msg[:c.pa.hdr])
if len(cir) == 0 {
return msg, false
}
+2 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2023 The NATS Authors
// Copyright 2012-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
@@ -55,7 +55,7 @@ func init() {
const (
// VERSION is the current version for the server.
VERSION = "2.10.25"
VERSION = "2.10.26"
// PROTO is the currently supported protocol.
// 0 was the original
+71 -49
View File
@@ -1,4 +1,4 @@
// Copyright 2019-2024 The NATS Authors
// Copyright 2019-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
@@ -60,7 +60,6 @@ type ConsumerInfo struct {
}
type ConsumerConfig struct {
// Durable is deprecated. All consumers should have names, picked by clients.
Durable string `json:"durable_name,omitempty"`
Name string `json:"name,omitempty"`
Description string `json:"description,omitempty"`
@@ -395,7 +394,7 @@ type consumer struct {
ackMsgs *ipQueue[*jsAckMsg]
// for stream signaling when multiple filters are set.
sigSubs []*subscription
sigSubs []string
}
// A single subject filter.
@@ -1583,6 +1582,12 @@ var (
// deleteNotActive must only be called from time.AfterFunc or in its own
// goroutine, as it can block on clean-up.
func (o *consumer) deleteNotActive() {
// Take a copy of these when the goroutine starts, mostly it avoids a
// race condition with tests that modify these consts, such as
// TestJetStreamClusterGhostEphemeralsAfterRestart.
cnaMax := consumerNotActiveMaxInterval
cnaStart := consumerNotActiveStartInterval
o.mu.Lock()
if o.mset == nil {
o.mu.Unlock()
@@ -1626,10 +1631,10 @@ func (o *consumer) deleteNotActive() {
if o.srv != nil {
qch = o.srv.quitCh
}
if o.js != nil {
cqch = o.js.clusterQuitC()
}
o.mu.Unlock()
if js != nil {
cqch = js.clusterQuitC()
}
// Useful for pprof.
setGoRoutineLabels(pprofLabels{
@@ -1663,8 +1668,8 @@ func (o *consumer) deleteNotActive() {
if ca != nil && cc != nil {
// Check to make sure we went away.
// Don't think this needs to be a monitored go routine.
jitter := time.Duration(rand.Int63n(int64(consumerNotActiveStartInterval)))
interval := consumerNotActiveStartInterval + jitter
jitter := time.Duration(rand.Int63n(int64(cnaStart)))
interval := cnaStart + jitter
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
@@ -1686,7 +1691,7 @@ func (o *consumer) deleteNotActive() {
if nca != nil && nca == ca {
s.Warnf("Consumer assignment for '%s > %s > %s' not cleaned up, retrying", acc, stream, name)
meta.ForwardProposal(removeEntry)
if interval < consumerNotActiveMaxInterval {
if interval < cnaMax {
interval *= 2
ticker.Reset(interval)
}
@@ -1744,11 +1749,16 @@ func (o *consumer) hasMaxDeliveries(seq uint64) bool {
if o.maxp > 0 && len(o.pending) >= o.maxp {
o.signalNewMessages()
}
// Cleanup our tracking.
delete(o.pending, seq)
if o.rdc != nil {
delete(o.rdc, seq)
// Make sure to remove from pending.
if p, ok := o.pending[seq]; ok && p != nil {
delete(o.pending, seq)
o.updateDelivered(p.Sequence, seq, dc, p.Timestamp)
}
// Ensure redelivered state is set, if not already.
if o.rdc == nil {
o.rdc = make(map[uint64]uint64)
}
o.rdc[seq] = dc
return true
}
return false
@@ -1859,9 +1869,6 @@ func (acc *Account) checkNewConsumerConfig(cfg, ncfg *ConsumerConfig) error {
if cfg.FlowControl != ncfg.FlowControl {
return errors.New("flow control can not be updated")
}
if cfg.MaxWaiting != ncfg.MaxWaiting {
return errors.New("max waiting can not be updated")
}
// Deliver Subject is conditional on if its bound.
if cfg.DeliverSubject != ncfg.DeliverSubject {
@@ -1876,6 +1883,10 @@ func (acc *Account) checkNewConsumerConfig(cfg, ncfg *ConsumerConfig) error {
}
}
if cfg.MaxWaiting != ncfg.MaxWaiting {
return errors.New("max waiting can not be updated")
}
// Check if BackOff is defined, MaxDeliver is within range.
if lbo := len(ncfg.BackOff); lbo > 0 && ncfg.MaxDeliver != -1 && lbo > ncfg.MaxDeliver {
return NewJSConsumerMaxDeliverBackoffError()
@@ -2806,17 +2817,20 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, reply string, doSample b
if sseq >= o.sseq {
// Let's make sure this is valid.
// This is only received on the consumer leader, so should never be higher
// than the last stream sequence.
// than the last stream sequence. But could happen if we've just become
// consumer leader, and we are not up-to-date on the stream yet.
var ss StreamState
mset.store.FastState(&ss)
if sseq > ss.LastSeq {
o.srv.Warnf("JetStream consumer '%s > %s > %s' ACK sequence %d past last stream sequence of %d",
o.acc.Name, o.stream, o.name, sseq, ss.LastSeq)
// FIXME(dlc) - For 2.11 onwards should we return an error here to the caller?
o.mu.Unlock()
return false
}
o.sseq = sseq + 1
// Even though another leader must have delivered a message with this sequence, we must not adjust
// the current pointer. This could otherwise result in a stuck consumer, where messages below this
// sequence can't be redelivered, and we'll have incorrect pending state and ack floors.
o.mu.Unlock()
return false
}
// Let the owning stream know if we are interest or workqueue retention based.
@@ -2979,6 +2993,7 @@ func (o *consumer) needAck(sseq uint64, subj string) bool {
var needAck bool
var asflr, osseq uint64
var pending map[uint64]*Pending
var rdc map[uint64]uint64
o.mu.RLock()
defer o.mu.RUnlock()
@@ -3003,7 +3018,7 @@ func (o *consumer) needAck(sseq uint64, subj string) bool {
}
if o.isLeader() {
asflr, osseq = o.asflr, o.sseq
pending = o.pending
pending, rdc = o.pending, o.rdc
} else {
if o.store == nil {
return false
@@ -3014,7 +3029,7 @@ func (o *consumer) needAck(sseq uint64, subj string) bool {
return sseq > o.asflr && !o.isFiltered()
}
// If loading state as here, the osseq is +1.
asflr, osseq, pending = state.AckFloor.Stream, state.Delivered.Stream+1, state.Pending
asflr, osseq, pending, rdc = state.AckFloor.Stream, state.Delivered.Stream+1, state.Pending, state.Redelivered
}
switch o.cfg.AckPolicy {
@@ -3030,6 +3045,12 @@ func (o *consumer) needAck(sseq uint64, subj string) bool {
}
}
// Finally check if redelivery of this message is tracked.
// If the message is not pending, it should be preserved if it reached max delivery.
if !needAck {
_, needAck = rdc[sseq]
}
return needAck
}
@@ -3488,7 +3509,10 @@ func (o *consumer) deliveryCount(seq uint64) uint64 {
if o.rdc == nil {
return 1
}
return o.rdc[seq]
if dc := o.rdc[seq]; dc >= 1 {
return dc
}
return 1
}
// Increase the delivery count for this message.
@@ -3802,10 +3826,7 @@ func (o *consumer) checkAckFloor() {
// Check if this message was pending.
o.mu.RLock()
p, isPending := o.pending[seq]
var rdc uint64 = 1
if o.rdc != nil {
rdc = o.rdc[seq]
}
rdc := o.deliveryCount(seq)
o.mu.RUnlock()
// If it was pending for us, get rid of it.
if isPending {
@@ -3823,10 +3844,7 @@ func (o *consumer) checkAckFloor() {
if p != nil {
dseq = p.Sequence
}
var rdc uint64 = 1
if o.rdc != nil {
rdc = o.rdc[seq]
}
rdc := o.deliveryCount(seq)
toTerm = append(toTerm, seq, dseq, rdc)
}
}
@@ -5400,10 +5418,7 @@ func (o *consumer) decStreamPending(sseq uint64, subj string) {
// Check if this message was pending.
p, wasPending := o.pending[sseq]
var rdc uint64 = 1
if o.rdc != nil {
rdc = o.rdc[sseq]
}
rdc := o.deliveryCount(sseq)
o.mu.Unlock()
@@ -5424,7 +5439,7 @@ func (o *consumer) account() *Account {
// Creates a sublist for consumer.
// All subjects share the same callback.
func (o *consumer) signalSubs() []*subscription {
func (o *consumer) signalSubs() []string {
o.mu.Lock()
defer o.mu.Unlock()
@@ -5432,15 +5447,15 @@ func (o *consumer) signalSubs() []*subscription {
return o.sigSubs
}
subs := []*subscription{}
if o.subjf == nil {
subs = append(subs, &subscription{subject: []byte(fwcs), icb: o.processStreamSignal})
if len(o.subjf) == 0 {
subs := []string{fwcs}
o.sigSubs = subs
return subs
}
subs := make([]string, 0, len(o.subjf))
for _, filter := range o.subjf {
subs = append(subs, &subscription{subject: []byte(filter.subject), icb: o.processStreamSignal})
subs = append(subs, filter.subject)
}
o.sigSubs = subs
return subs
@@ -5450,7 +5465,7 @@ func (o *consumer) signalSubs() []*subscription {
// We know that this subject matches us by how the parent handles registering us with the signaling sublist,
// but we must check if we are leader.
// We do need the sequence of the message however and we use the msg as the encoded seq.
func (o *consumer) processStreamSignal(_ *subscription, _ *client, _ *Account, subject, _ string, seqb []byte) {
func (o *consumer) processStreamSignal(seq uint64) {
// We can get called here now when not leader, so bail fast
// and without acquiring any locks.
if !o.leader.Load() {
@@ -5461,10 +5476,6 @@ func (o *consumer) processStreamSignal(_ *subscription, _ *client, _ *Account, s
if o.mset == nil {
return
}
var le = binary.LittleEndian
seq := le.Uint64(seqb)
if seq > o.npf {
o.npc++
}
@@ -5539,6 +5550,7 @@ func (o *consumer) isMonitorRunning() bool {
// If we detect that our ackfloor is higher than the stream's last sequence, return this error.
var errAckFloorHigherThanLastSeq = errors.New("consumer ack floor is higher than streams last sequence")
var errAckFloorInvalid = errors.New("consumer ack floor is invalid")
// If we are a consumer of an interest or workqueue policy stream, process that state and make sure consistent.
func (o *consumer) checkStateForInterestStream(ss *StreamState) error {
@@ -5568,7 +5580,7 @@ func (o *consumer) checkStateForInterestStream(ss *StreamState) error {
asflr := state.AckFloor.Stream
// Protect ourselves against rolling backwards.
if asflr&(1<<63) != 0 {
return nil
return errAckFloorInvalid
}
// Check if the underlying stream's last sequence is less than our floor.
@@ -5587,6 +5599,7 @@ func (o *consumer) checkStateForInterestStream(ss *StreamState) error {
fseq = chkfloor
}
var retryAsflr uint64
for seq = fseq; asflr > 0 && seq <= asflr; seq++ {
if filters != nil {
_, nseq, err = store.LoadNextMsgMulti(filters, seq, &smv)
@@ -5599,15 +5612,24 @@ func (o *consumer) checkStateForInterestStream(ss *StreamState) error {
}
// Only ack though if no error and seq <= ack floor.
if err == nil && seq <= asflr {
mset.ackMsg(o, seq)
didRemove := mset.ackMsg(o, seq)
// Removing the message could fail. For example if we're behind on stream applies.
// Overwrite retry floor (only the first time) to allow us to check next time if the removal was successful.
if didRemove && retryAsflr == 0 {
retryAsflr = seq
}
}
}
// If retry floor was not overwritten, set to ack floor+1, we don't need to account for any retries below it.
if retryAsflr == 0 {
retryAsflr = asflr + 1
}
o.mu.Lock()
// Update our check floor.
// Check floor must never be greater than ack floor+1, otherwise subsequent calls to this function would skip work.
if asflr+1 > o.chkflr {
o.chkflr = asflr + 1
if retryAsflr > o.chkflr {
o.chkflr = retryAsflr
}
// See if we need to process this update if our parent stream is not a limits policy stream.
state, _ = o.store.State()
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2021 The NATS Authors
// Copyright 2012-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2022 The NATS Authors
// Copyright 2022-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2020 The NATS Authors
// Copyright 2020-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2021 The NATS Authors
// Copyright 2012-2024 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
+35 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2018-2023 The NATS Authors
// Copyright 2018-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
@@ -1215,6 +1215,14 @@ func (s *Server) initEventTracking() {
optz := &ExpvarzEventOptions{}
s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.expvarz(optz), nil })
},
"IPQUEUESZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
optz := &IpqueueszEventOptions{}
s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.Ipqueuesz(&optz.IpqueueszOptions), nil })
},
"RAFTZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
optz := &RaftzEventOptions{}
s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.Raftz(&optz.RaftzOptions), nil })
},
}
profilez := func(_ *subscription, c *client, _ *Account, _, rply string, rmsg []byte) {
hdr, msg := c.msgParts(rmsg)
@@ -1921,6 +1929,18 @@ type ExpvarzEventOptions struct {
EventFilterOptions
}
// In the context of system events, IpqueueszEventOptions are options passed to Ipqueuesz
type IpqueueszEventOptions struct {
EventFilterOptions
IpqueueszOptions
}
// In the context of system events, RaftzEventOptions are options passed to Raftz
type RaftzEventOptions struct {
EventFilterOptions
RaftzOptions
}
// returns true if the request does NOT apply to this server and can be ignored.
// DO NOT hold the server lock when
func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
@@ -2043,6 +2063,20 @@ type ServerAPIExpvarzResponse struct {
Error *ApiError `json:"error,omitempty"`
}
// ServerAPIpqueueszResponse is the response type for ipqueuesz
type ServerAPIpqueueszResponse struct {
Server *ServerInfo `json:"server"`
Data *IpqueueszStatus `json:"data,omitempty"`
Error *ApiError `json:"error,omitempty"`
}
// ServerAPIRaftzResponse is the response type for raftz
type ServerAPIRaftzResponse struct {
Server *ServerInfo `json:"server"`
Data *RaftzStatus `json:"data,omitempty"`
Error *ApiError `json:"error,omitempty"`
}
// statszReq is a request for us to respond with current statsz.
func (s *Server) statszReq(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
if !s.EventsEnabled() {
+242 -42
View File
@@ -1,4 +1,4 @@
// Copyright 2019-2024 The NATS Authors
// Copyright 2019-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
@@ -33,6 +33,7 @@ import (
"net"
"os"
"path/filepath"
"runtime"
"slices"
"sort"
"strings"
@@ -1431,15 +1432,14 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, []uint64, error) {
if seq == 0 || seq&ebit != 0 || seq < fseq {
seq = seq &^ ebit
if seq >= fseq {
// Only add to dmap if past recorded first seq and non-zero.
if seq != 0 {
addToDmap(seq)
}
atomic.StoreUint64(&mb.last.seq, seq)
mb.last.ts = ts
if mb.msgs == 0 {
atomic.StoreUint64(&mb.first.seq, seq+1)
mb.first.ts = 0
} else if seq != 0 {
// Only add to dmap if past recorded first seq and non-zero.
addToDmap(seq)
}
}
index += rl
@@ -2004,7 +2004,7 @@ func (fs *fileStore) expireMsgsOnRecover() error {
}
// Make sure we do subject cleanup as well.
mb.ensurePerSubjectInfoLoaded()
mb.fss.Iter(func(bsubj []byte, ss *SimpleState) bool {
mb.fss.IterOrdered(func(bsubj []byte, ss *SimpleState) bool {
subj := bytesToString(bsubj)
for i := uint64(0); i < ss.Msgs; i++ {
fs.removePerSubject(subj)
@@ -2207,12 +2207,15 @@ func (fs *fileStore) GetSeqFromTime(t time.Time) uint64 {
// Find the first matching message against a sublist.
func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg) (*StoreMsg, bool, error) {
mb.mu.Lock()
defer mb.mu.Unlock()
// Will just do linear walk for now.
// TODO(dlc) - Be better at skipping blocks that will not match us regardless.
var didLoad bool
var updateLLTS bool
defer func() {
if updateLLTS {
mb.llts = time.Now().UnixNano()
}
mb.mu.Unlock()
}()
// Need messages loaded from here on out.
if mb.cacheNotLoaded() {
if err := mb.loadMsgsWithLock(); err != nil {
@@ -2231,20 +2234,88 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
sm = new(StoreMsg)
}
for seq := start; seq <= lseq; seq++ {
llseq := mb.llseq
fsm, err := mb.cacheLookup(seq, sm)
if err != nil {
continue
// If the FSS state has fewer entries than sequences in the linear scan,
// then use intersection instead as likely going to be cheaper. This will
// often be the case with high numbers of deletes, as well as a smaller
// number of subjects in the block.
if uint64(mb.fss.Size()) < lseq-start {
// If there are no subject matches then this is effectively no-op.
hseq := uint64(math.MaxUint64)
IntersectStree(mb.fss, sl, func(subj []byte, ss *SimpleState) {
if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
// mb is already loaded into the cache so should be fast-ish.
mb.recalculateForSubj(bytesToString(subj), ss)
}
first := ss.First
if start > first {
first = start
}
if first > ss.Last || first >= hseq {
// The start cutoff is after the last sequence for this subject,
// or we think we already know of a subject with an earlier msg
// than our first seq for this subject.
return
}
if first == ss.First {
// If the start floor is below where this subject starts then we can
// short-circuit, avoiding needing to scan for the next message.
if fsm, err := mb.cacheLookup(ss.First, sm); err == nil {
sm = fsm
hseq = ss.First
}
return
}
for seq := first; seq <= ss.Last; seq++ {
// Otherwise we have a start floor that intersects where this subject
// has messages in the block, so we need to walk up until we find a
// message matching the subject.
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
// Instead we will update it only once in a defer.
updateLLTS = true
continue
}
llseq := mb.llseq
fsm, err := mb.cacheLookup(seq, sm)
if err != nil {
continue
}
updateLLTS = false // cacheLookup already updated it.
if sl.HasInterest(fsm.subj) {
hseq = seq
sm = fsm
break
}
// If we are here we did not match, so put the llseq back.
mb.llseq = llseq
}
})
if hseq < uint64(math.MaxUint64) && sm != nil {
return sm, didLoad, nil
}
expireOk := seq == lseq && mb.llseq == seq
if sl.HasInterest(fsm.subj) {
return fsm, expireOk, nil
} else {
for seq := start; seq <= lseq; seq++ {
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
// Instead we will update it only once in a defer.
updateLLTS = true
continue
}
llseq := mb.llseq
fsm, err := mb.cacheLookup(seq, sm)
if err != nil {
continue
}
expireOk := seq == lseq && mb.llseq == seq
updateLLTS = false // cacheLookup already updated it.
if sl.HasInterest(fsm.subj) {
return fsm, expireOk, nil
}
// If we are here we did not match, so put the llseq back.
mb.llseq = llseq
}
// If we are here we did not match, so put the llseq back.
mb.llseq = llseq
}
return nil, didLoad, ErrStoreMsgNotFound
}
@@ -2252,7 +2323,13 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
// fs lock should be held.
func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *StoreMsg) (*StoreMsg, bool, error) {
mb.mu.Lock()
defer mb.mu.Unlock()
var updateLLTS bool
defer func() {
if updateLLTS {
mb.llts = time.Now().UnixNano()
}
mb.mu.Unlock()
}()
fseq, isAll, subs := start, filter == _EMPTY_ || filter == fwcs, []string{filter}
@@ -2364,6 +2441,12 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
}
for seq := fseq; seq <= lseq; seq++ {
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
// Instead we will update it only once in a defer.
updateLLTS = true
continue
}
llseq := mb.llseq
fsm, err := mb.cacheLookup(seq, sm)
if err != nil {
@@ -2372,6 +2455,7 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
}
continue
}
updateLLTS = false // cacheLookup already updated it.
expireOk := seq == lseq && mb.llseq == seq
if isAll {
return fsm, expireOk, nil
@@ -2876,6 +2960,7 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
// Now check if we need to inspect the seqStart block.
// Grab write lock in case we need to load in msgs.
mb.mu.Lock()
var updateLLTS bool
var shouldExpire bool
// We need to walk this block to correct accounting from above.
if sseq > mb.first.seq {
@@ -2889,10 +2974,16 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
}
var smv StoreMsg
for seq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq); seq <= lseq; seq++ {
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
updateLLTS = true
continue
}
sm, _ := mb.cacheLookup(seq, &smv)
if sm == nil || sm.subj == _EMPTY_ || !lbm[sm.subj] {
continue
}
updateLLTS = false // cacheLookup already updated it.
if isMatch(sm.subj) {
// If less than sseq adjust off of total as long as this subject matched the last block.
if seq < sseq {
@@ -2913,6 +3004,9 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
if shouldExpire {
mb.tryForceExpireCacheLocked()
}
if updateLLTS {
mb.llts = time.Now().UnixNano()
}
mb.mu.Unlock()
return total, validThrough
}
@@ -3023,6 +3117,7 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
}
// We need to scan this block.
var shouldExpire bool
var updateLLTS bool
mb.mu.Lock()
// Check if we should include all of this block in adjusting. If so work with metadata.
if sseq > atomic.LoadUint64(&mb.last.seq) {
@@ -3055,10 +3150,16 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
// We need to walk all messages in this block
var smv StoreMsg
for seq := atomic.LoadUint64(&mb.first.seq); seq < last; seq++ {
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
updateLLTS = true
continue
}
sm, _ := mb.cacheLookup(seq, &smv)
if sm == nil || sm.subj == _EMPTY_ {
continue
}
updateLLTS = false // cacheLookup already updated it.
// Check if it matches our filter.
if sm.seq < sseq && isMatch(sm.subj) {
adjust++
@@ -3069,6 +3170,9 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
if shouldExpire {
mb.tryForceExpireCacheLocked()
}
if updateLLTS {
mb.llts = time.Now().UnixNano()
}
mb.mu.Unlock()
}
// Make final adjustment.
@@ -3109,7 +3213,7 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
// See if filter was provided but its the only subject.
if !isAll && fs.psim.Size() == 1 {
fs.psim.Iter(func(subject []byte, _ *psi) bool {
fs.psim.IterFast(func(subject []byte, _ *psi) bool {
isAll = sl.HasInterest(bytesToString(subject))
return true
})
@@ -3166,6 +3270,7 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
// Grab write lock in case we need to load in msgs.
mb.mu.Lock()
var shouldExpire bool
var updateLLTS bool
// We need to walk this block to correct accounting from above.
if sseq > mb.first.seq {
// Track the ones we add back in case more than one.
@@ -3178,10 +3283,16 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
}
var smv StoreMsg
for seq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq); seq <= lseq; seq++ {
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
updateLLTS = true
continue
}
sm, _ := mb.cacheLookup(seq, &smv)
if sm == nil || sm.subj == _EMPTY_ || !lbm[sm.subj] {
continue
}
updateLLTS = false // cacheLookup already updated it.
if isMatch(sm.subj) {
// If less than sseq adjust off of total as long as this subject matched the last block.
if seq < sseq {
@@ -3202,6 +3313,9 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
if shouldExpire {
mb.tryForceExpireCacheLocked()
}
if updateLLTS {
mb.llts = time.Now().UnixNano()
}
mb.mu.Unlock()
return total, validThrough
}
@@ -3229,6 +3343,7 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
var t uint64
var havePartial bool
var updateLLTS bool
IntersectStree[SimpleState](mb.fss, sl, func(bsubj []byte, ss *SimpleState) {
subj := bytesToString(bsubj)
if havePartial {
@@ -3261,8 +3376,14 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
}
var smv StoreMsg
for seq, lseq := start, atomic.LoadUint64(&mb.last.seq); seq <= lseq; seq++ {
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
updateLLTS = true
continue
}
if sm, _ := mb.cacheLookup(seq, &smv); sm != nil && isMatch(sm.subj) {
t++
updateLLTS = false // cacheLookup already updated it.
}
}
}
@@ -3270,6 +3391,9 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
if shouldExpire {
mb.tryForceExpireCacheLocked()
}
if updateLLTS {
mb.llts = time.Now().UnixNano()
}
mb.mu.Unlock()
total += t
}
@@ -3314,6 +3438,7 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
}
// We need to scan this block.
var shouldExpire bool
var updateLLTS bool
mb.mu.Lock()
// Check if we should include all of this block in adjusting. If so work with metadata.
if sseq > atomic.LoadUint64(&mb.last.seq) {
@@ -3345,10 +3470,16 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
// We need to walk all messages in this block
var smv StoreMsg
for seq := atomic.LoadUint64(&mb.first.seq); seq < last; seq++ {
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
updateLLTS = true
continue
}
sm, _ := mb.cacheLookup(seq, &smv)
if sm == nil || sm.subj == _EMPTY_ {
continue
}
updateLLTS = false // cacheLookup already updated it.
// Check if it matches our filter.
if sm.seq < sseq && isMatch(sm.subj) {
adjust++
@@ -3359,6 +3490,9 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
if shouldExpire {
mb.tryForceExpireCacheLocked()
}
if updateLLTS {
mb.llts = time.Now().UnixNano()
}
mb.mu.Unlock()
}
// Make final adjustment.
@@ -3992,7 +4126,7 @@ func (fs *fileStore) enforceMsgPerSubjectLimit(fireCallback bool) {
// collect all that are not correct.
needAttention := make(map[string]*psi)
fs.psim.Iter(func(subj []byte, psi *psi) bool {
fs.psim.IterFast(func(subj []byte, psi *psi) bool {
numMsgs += psi.total
if psi.total > maxMsgsPer {
needAttention[string(subj)] = psi
@@ -4017,7 +4151,7 @@ func (fs *fileStore) enforceMsgPerSubjectLimit(fireCallback bool) {
fs.rebuildStateLocked(nil)
// Need to redo blocks that need attention.
needAttention = make(map[string]*psi)
fs.psim.Iter(func(subj []byte, psi *psi) bool {
fs.psim.IterFast(func(subj []byte, psi *psi) bool {
if psi.total > maxMsgsPer {
needAttention[string(subj)] = psi
}
@@ -5174,6 +5308,7 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
if ss, ok := mb.fss.Find(stringToBytes(subj)); ok && ss != nil {
ss.Msgs++
ss.Last = seq
ss.lastNeedsUpdate = false
} else {
mb.fss.Insert(stringToBytes(subj), SimpleState{Msgs: 1, First: seq, Last: seq})
}
@@ -5188,9 +5323,7 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
// With headers, high bit on total length will be set.
// total_len(4) sequence(8) timestamp(8) subj_len(2) subj hdr_len(4) hdr msg hash(8)
// First write header, etc.
var le = binary.LittleEndian
var hdr [msgHdrSize]byte
l := uint32(rl)
hasHeaders := len(mhdr) > 0
@@ -5198,13 +5331,15 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
l |= hbit
}
// Reserve space for the header on the underlying buffer.
mb.cache.buf = append(mb.cache.buf, make([]byte, msgHdrSize)...)
hdr := mb.cache.buf[len(mb.cache.buf)-msgHdrSize : len(mb.cache.buf)]
le.PutUint32(hdr[0:], l)
le.PutUint64(hdr[4:], seq)
le.PutUint64(hdr[12:], uint64(ts))
le.PutUint16(hdr[20:], uint16(len(subj)))
// Now write to underlying buffer.
mb.cache.buf = append(mb.cache.buf, hdr[:]...)
mb.cache.buf = append(mb.cache.buf, subj...)
if hasHeaders {
@@ -5218,13 +5353,12 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
// Calculate hash.
mb.hh.Reset()
mb.hh.Write(hdr[4:20])
mb.hh.Write([]byte(subj))
mb.hh.Write(stringToBytes(subj))
if hasHeaders {
mb.hh.Write(mhdr)
}
mb.hh.Write(msg)
checksum := mb.hh.Sum(nil)
// Grab last checksum
checksum := mb.hh.Sum(mb.lchk[:0:highwayhash.Size64])
copy(mb.lchk[0:], checksum)
// Update write through cache.
@@ -5896,6 +6030,7 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
if ss, ok := mb.fss.Find(bsubj); ok && ss != nil {
ss.Msgs++
ss.Last = seq
ss.lastNeedsUpdate = false
} else {
mb.fss.Insert(bsubj, SimpleState{
Msgs: 1,
@@ -6763,6 +6898,57 @@ func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64, sm *Store
return nil, fs.state.LastSeq, ErrStoreEOF
}
// Will load the next non-deleted msg starting at the start sequence and walking backwards.
func (fs *fileStore) LoadPrevMsg(start uint64, smp *StoreMsg) (sm *StoreMsg, err error) {
fs.mu.RLock()
defer fs.mu.RUnlock()
if fs.closed {
return nil, ErrStoreClosed
}
if fs.state.Msgs == 0 || start < fs.state.FirstSeq {
return nil, ErrStoreEOF
}
if start > fs.state.LastSeq {
start = fs.state.LastSeq
}
if smp == nil {
smp = new(StoreMsg)
}
if bi, _ := fs.selectMsgBlockWithIndex(start); bi >= 0 {
for i := bi; i >= 0; i-- {
mb := fs.blks[i]
mb.mu.Lock()
// Need messages loaded from here on out.
if mb.cacheNotLoaded() {
if err := mb.loadMsgsWithLock(); err != nil {
mb.mu.Unlock()
return nil, err
}
}
lseq, fseq := atomic.LoadUint64(&mb.last.seq), atomic.LoadUint64(&mb.first.seq)
if start > lseq {
start = lseq
}
for seq := start; seq >= fseq; seq-- {
if mb.dmap.Exists(seq) {
continue
}
if sm, err := mb.cacheLookup(seq, smp); err == nil {
mb.mu.Unlock()
return sm, nil
}
}
mb.mu.Unlock()
}
}
return nil, ErrStoreEOF
}
// Type returns the type of the underlying store.
func (fs *fileStore) Type() StorageType {
return FileStorage
@@ -6831,11 +7017,7 @@ func (fs *fileStore) State() StreamState {
}
// Add in deleted.
mb.dmap.Range(func(seq uint64) bool {
if seq < fseq {
mb.dmap.Delete(seq)
} else {
state.Deleted = append(state.Deleted, seq)
}
state.Deleted = append(state.Deleted, seq)
return true
})
mb.mu.Unlock()
@@ -7353,7 +7535,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
bytes += mb.bytes
// Make sure we do subject cleanup as well.
mb.ensurePerSubjectInfoLoaded()
mb.fss.Iter(func(bsubj []byte, ss *SimpleState) bool {
mb.fss.IterOrdered(func(bsubj []byte, ss *SimpleState) bool {
subj := bytesToString(bsubj)
for i := uint64(0); i < ss.Msgs; i++ {
fs.removePerSubject(subj)
@@ -7387,7 +7569,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
if err == errDeletedMsg {
// Update dmap.
if !smb.dmap.IsEmpty() {
smb.dmap.Delete(seq)
smb.dmap.Delete(mseq)
}
} else if sm != nil {
sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
@@ -7876,8 +8058,11 @@ func (mb *msgBlock) recalculateForSubj(subj string, ss *SimpleState) {
}
if startSlot >= len(mb.cache.idx) {
ss.First = ss.Last
ss.firstNeedsUpdate = false
ss.lastNeedsUpdate = false
return
}
endSlot := int(ss.Last - mb.cache.fseq)
if endSlot < 0 {
endSlot = 0
@@ -7904,6 +8089,8 @@ func (mb *msgBlock) recalculateForSubj(subj string, ss *SimpleState) {
li := int(bi) - mb.cache.off
if li >= len(mb.cache.buf) {
ss.First = ss.Last
// Only need to reset ss.lastNeedsUpdate, ss.firstNeedsUpdate is already reset above.
ss.lastNeedsUpdate = false
return
}
buf := mb.cache.buf[li:]
@@ -8007,6 +8194,11 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
var smv StoreMsg
fseq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq)
for seq := fseq; seq <= lseq; seq++ {
if mb.dmap.Exists(seq) {
// Optimisation to avoid calling cacheLookup which hits time.Now().
// It gets set later on if the fss is non-empty anyway.
continue
}
sm, err := mb.cacheLookup(seq, &smv)
if err != nil {
// Since we are walking by sequence we can ignore some errors that are benign to rebuilding our state.
@@ -8022,6 +8214,7 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
if ss, ok := mb.fss.Find(stringToBytes(sm.subj)); ok && ss != nil {
ss.Msgs++
ss.Last = seq
ss.lastNeedsUpdate = false
} else {
mb.fss.Insert(stringToBytes(sm.subj), SimpleState{Msgs: 1, First: seq, Last: seq})
}
@@ -8066,7 +8259,7 @@ func (fs *fileStore) populateGlobalPerSubjectInfo(mb *msgBlock) {
}
// Now populate psim.
mb.fss.Iter(func(bsubj []byte, ss *SimpleState) bool {
mb.fss.IterFast(func(bsubj []byte, ss *SimpleState) bool {
if len(bsubj) > 0 {
if info, ok := fs.psim.Find(bsubj); ok {
info.total += ss.Msgs
@@ -9482,8 +9675,15 @@ var dios chan struct{}
// Used to setup our simplistic counting semaphore using buffered channels.
// golang.org's semaphore seemed a bit heavy.
func init() {
// Limit ourselves to a max of 4 blocking IO calls.
const nIO = 4
// Limit ourselves to a sensible number of blocking I/O calls. Range between
// 4-16 concurrent disk I/Os based on CPU cores, or 50% of cores if greater
// than 32 cores.
mp := runtime.GOMAXPROCS(-1)
nIO := min(16, max(4, mp))
if mp > 32 {
// If the system has more than 32 cores then limit dios to 50% of cores.
nIO = max(16, min(mp, mp/2))
}
dios = make(chan struct{}, nIO)
// Fill it up to start.
for i := 0; i < nIO; i++ {
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2020 The NATS Authors
// Copyright 2020-2022 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
+23 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2018-2023 The NATS Authors
// Copyright 2018-2024 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
@@ -2499,8 +2499,13 @@ var subPool = &sync.Pool{
// that the message is not sent to a given gateway if for instance
// it is known that this gateway has no interest in the account or
// subject, etc..
// When invoked from a LEAF connection, `checkLeafQF` should be passed as `true`
// so that we skip any queue subscription interest that is not part of the
// `c.pa.queues` filter (similar to what we do in `processMsgResults`). However,
// when processing service imports, then this boolean should be passes as `false`,
// regardless if it is a LEAF connection or not.
// <Invoked from any client connection's readLoop>
func (c *client) sendMsgToGateways(acc *Account, msg, subject, reply []byte, qgroups [][]byte) bool {
func (c *client) sendMsgToGateways(acc *Account, msg, subject, reply []byte, qgroups [][]byte, checkLeafQF bool) bool {
// We had some times when we were sending across a GW with no subject, and the other side would break
// due to parser error. These need to be fixed upstream but also double check here.
if len(subject) == 0 {
@@ -2577,6 +2582,21 @@ func (c *client) sendMsgToGateways(acc *Account, msg, subject, reply []byte, qgr
qsubs := qr.qsubs[i]
if len(qsubs) > 0 {
queue := qsubs[0].queue
if checkLeafQF {
// Skip any queue that is not in the leaf's queue filter.
skip := true
for _, qn := range c.pa.queues {
if bytes.Equal(queue, qn) {
skip = false
break
}
}
if skip {
continue
}
// Now we still need to check that it was not delivered
// locally by checking the given `qgroups`.
}
add := true
for _, qn := range qgroups {
if bytes.Equal(queue, qn) {
@@ -2969,7 +2989,7 @@ func (c *client) handleGatewayReply(msg []byte) (processed bool) {
// we now need to send the message with the real subject to
// gateways in case they have interest on that reply subject.
if !isServiceReply {
c.sendMsgToGateways(acc, msg, c.pa.subject, c.pa.reply, queues)
c.sendMsgToGateways(acc, msg, c.pa.subject, c.pa.reply, queues, false)
}
} else if c.kind == GATEWAY {
// Only if we are a gateway connection should we try to route
+532
View File
@@ -0,0 +1,532 @@
// Copyright 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
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package gsl
import (
"errors"
"sync"
"github.com/nats-io/nats-server/v2/server/stree"
)
// Sublist is a routing mechanism to handle subject distribution and
// provides a facility to match subjects from published messages to
// interested subscribers. Subscribers can have wildcard subjects to
// match multiple published subjects.
// Common byte variables for wildcards and token separator.
const (
pwc = '*'
pwcs = "*"
fwc = '>'
fwcs = ">"
tsep = "."
btsep = '.'
_EMPTY_ = ""
)
// Sublist related errors
var (
ErrInvalidSubject = errors.New("gsl: invalid subject")
ErrNotFound = errors.New("gsl: no matches found")
ErrNilChan = errors.New("gsl: nil channel")
ErrAlreadyRegistered = errors.New("gsl: notification already registered")
)
// A GenericSublist stores and efficiently retrieves subscriptions.
type GenericSublist[T comparable] struct {
sync.RWMutex
root *level[T]
count uint32
}
// A node contains subscriptions and a pointer to the next level.
type node[T comparable] struct {
next *level[T]
subs map[T]string // value -> subject
}
// A level represents a group of nodes and special pointers to
// wildcard nodes.
type level[T comparable] struct {
nodes map[string]*node[T]
pwc, fwc *node[T]
}
// Create a new default node.
func newNode[T comparable]() *node[T] {
return &node[T]{subs: make(map[T]string)}
}
// Create a new default level.
func newLevel[T comparable]() *level[T] {
return &level[T]{nodes: make(map[string]*node[T])}
}
// NewSublist will create a default sublist with caching enabled per the flag.
func NewSublist[T comparable]() *GenericSublist[T] {
return &GenericSublist[T]{root: newLevel[T]()}
}
// Insert adds a subscription into the sublist
func (s *GenericSublist[T]) Insert(subject string, value T) error {
tsa := [32]string{}
tokens := tsa[:0]
start := 0
for i := 0; i < len(subject); i++ {
if subject[i] == btsep {
tokens = append(tokens, subject[start:i])
start = i + 1
}
}
tokens = append(tokens, subject[start:])
s.Lock()
var sfwc bool
var n *node[T]
l := s.root
for _, t := range tokens {
lt := len(t)
if lt == 0 || sfwc {
s.Unlock()
return ErrInvalidSubject
}
if lt > 1 {
n = l.nodes[t]
} else {
switch t[0] {
case pwc:
n = l.pwc
case fwc:
n = l.fwc
sfwc = true
default:
n = l.nodes[t]
}
}
if n == nil {
n = newNode[T]()
if lt > 1 {
l.nodes[t] = n
} else {
switch t[0] {
case pwc:
l.pwc = n
case fwc:
l.fwc = n
default:
l.nodes[t] = n
}
}
}
if n.next == nil {
n.next = newLevel[T]()
}
l = n.next
}
n.subs[value] = subject
s.count++
s.Unlock()
return nil
}
// Match will match all entries to the literal subject.
// It will return a set of results for both normal and queue subscribers.
func (s *GenericSublist[T]) Match(subject string, cb func(T)) {
s.match(subject, cb, true)
}
// MatchBytes will match all entries to the literal subject.
// It will return a set of results for both normal and queue subscribers.
func (s *GenericSublist[T]) MatchBytes(subject []byte, cb func(T)) {
s.match(string(subject), cb, true)
}
// HasInterest will return whether or not there is any interest in the subject.
// In cases where more detail is not required, this may be faster than Match.
func (s *GenericSublist[T]) HasInterest(subject string) bool {
return s.hasInterest(subject, true, nil)
}
// NumInterest will return the number of subs interested in the subject.
// In cases where more detail is not required, this may be faster than Match.
func (s *GenericSublist[T]) NumInterest(subject string) (np int) {
s.hasInterest(subject, true, &np)
return
}
func (s *GenericSublist[T]) match(subject string, cb func(T), doLock bool) {
tsa := [32]string{}
tokens := tsa[:0]
start := 0
for i := 0; i < len(subject); i++ {
if subject[i] == btsep {
if i-start == 0 {
return
}
tokens = append(tokens, subject[start:i])
start = i + 1
}
}
if start >= len(subject) {
return
}
tokens = append(tokens, subject[start:])
if doLock {
s.RLock()
defer s.RUnlock()
}
matchLevel(s.root, tokens, cb)
}
func (s *GenericSublist[T]) hasInterest(subject string, doLock bool, np *int) bool {
tsa := [32]string{}
tokens := tsa[:0]
start := 0
for i := 0; i < len(subject); i++ {
if subject[i] == btsep {
if i-start == 0 {
return false
}
tokens = append(tokens, subject[start:i])
start = i + 1
}
}
if start >= len(subject) {
return false
}
tokens = append(tokens, subject[start:])
if doLock {
s.RLock()
defer s.RUnlock()
}
return matchLevelForAny(s.root, tokens, np)
}
func matchLevelForAny[T comparable](l *level[T], toks []string, np *int) bool {
var pwc, n *node[T]
for i, t := range toks {
if l == nil {
return false
}
if l.fwc != nil {
if np != nil {
*np += len(l.fwc.subs)
}
return true
}
if pwc = l.pwc; pwc != nil {
if match := matchLevelForAny(pwc.next, toks[i+1:], np); match {
return true
}
}
n = l.nodes[t]
if n != nil {
l = n.next
} else {
l = nil
}
}
if n != nil {
if np != nil {
*np += len(n.subs)
}
return len(n.subs) > 0
}
if pwc != nil {
if np != nil {
*np += len(pwc.subs)
}
return len(pwc.subs) > 0
}
return false
}
// callbacksForResults will make the necessary callbacks for each
// result in this node.
func callbacksForResults[T comparable](n *node[T], cb func(T)) {
for sub := range n.subs {
cb(sub)
}
}
// matchLevel is used to recursively descend into the trie.
func matchLevel[T comparable](l *level[T], toks []string, cb func(T)) {
var pwc, n *node[T]
for i, t := range toks {
if l == nil {
return
}
if l.fwc != nil {
callbacksForResults(l.fwc, cb)
}
if pwc = l.pwc; pwc != nil {
matchLevel(pwc.next, toks[i+1:], cb)
}
n = l.nodes[t]
if n != nil {
l = n.next
} else {
l = nil
}
}
if n != nil {
callbacksForResults(n, cb)
}
if pwc != nil {
callbacksForResults(pwc, cb)
}
}
// lnt is used to track descent into levels for a removal for pruning.
type lnt[T comparable] struct {
l *level[T]
n *node[T]
t string
}
// Raw low level remove, can do batches with lock held outside.
func (s *GenericSublist[T]) remove(subject string, value T, shouldLock bool) error {
tsa := [32]string{}
tokens := tsa[:0]
start := 0
for i := 0; i < len(subject); i++ {
if subject[i] == btsep {
tokens = append(tokens, subject[start:i])
start = i + 1
}
}
tokens = append(tokens, subject[start:])
if shouldLock {
s.Lock()
defer s.Unlock()
}
var sfwc bool
var n *node[T]
l := s.root
// Track levels for pruning
var lnts [32]lnt[T]
levels := lnts[:0]
for _, t := range tokens {
lt := len(t)
if lt == 0 || sfwc {
return ErrInvalidSubject
}
if l == nil {
return ErrNotFound
}
if lt > 1 {
n = l.nodes[t]
} else {
switch t[0] {
case pwc:
n = l.pwc
case fwc:
n = l.fwc
sfwc = true
default:
n = l.nodes[t]
}
}
if n != nil {
levels = append(levels, lnt[T]{l, n, t})
l = n.next
} else {
l = nil
}
}
if !s.removeFromNode(n, value) {
return ErrNotFound
}
s.count--
for i := len(levels) - 1; i >= 0; i-- {
l, n, t := levels[i].l, levels[i].n, levels[i].t
if n.isEmpty() {
l.pruneNode(n, t)
}
}
return nil
}
// Remove will remove a subscription.
func (s *GenericSublist[T]) Remove(subject string, value T) error {
return s.remove(subject, value, true)
}
// pruneNode is used to prune an empty node from the tree.
func (l *level[T]) pruneNode(n *node[T], t string) {
if n == nil {
return
}
if n == l.fwc {
l.fwc = nil
} else if n == l.pwc {
l.pwc = nil
} else {
delete(l.nodes, t)
}
}
// isEmpty will test if the node has any entries. Used
// in pruning.
func (n *node[T]) isEmpty() bool {
return len(n.subs) == 0 && (n.next == nil || n.next.numNodes() == 0)
}
// Return the number of nodes for the given level.
func (l *level[T]) numNodes() int {
num := len(l.nodes)
if l.pwc != nil {
num++
}
if l.fwc != nil {
num++
}
return num
}
// Remove the sub for the given node.
func (s *GenericSublist[T]) removeFromNode(n *node[T], value T) (found bool) {
if n == nil {
return false
}
if _, found = n.subs[value]; found {
delete(n.subs, value)
}
return found
}
// Count returns the number of subscriptions.
func (s *GenericSublist[T]) Count() uint32 {
s.RLock()
defer s.RUnlock()
return s.count
}
// numLevels will return the maximum number of levels
// contained in the Sublist tree.
func (s *GenericSublist[T]) numLevels() int {
return visitLevel(s.root, 0)
}
// visitLevel is used to descend the Sublist tree structure
// recursively.
func visitLevel[T comparable](l *level[T], depth int) int {
if l == nil || l.numNodes() == 0 {
return depth
}
depth++
maxDepth := depth
for _, n := range l.nodes {
if n == nil {
continue
}
newDepth := visitLevel(n.next, depth)
if newDepth > maxDepth {
maxDepth = newDepth
}
}
if l.pwc != nil {
pwcDepth := visitLevel(l.pwc.next, depth)
if pwcDepth > maxDepth {
maxDepth = pwcDepth
}
}
if l.fwc != nil {
fwcDepth := visitLevel(l.fwc.next, depth)
if fwcDepth > maxDepth {
maxDepth = fwcDepth
}
}
return maxDepth
}
// IntersectStree will match all items in the given subject tree that
// have interest expressed in the given sublist. The callback will only be called
// once for each subject, regardless of overlapping subscriptions in the sublist.
func IntersectStree[T1 any, T2 comparable](st *stree.SubjectTree[T1], sl *GenericSublist[T2], cb func(subj []byte, entry *T1)) {
var _subj [255]byte
intersectStree(st, sl.root, _subj[:0], cb)
}
func intersectStree[T1 any, T2 comparable](st *stree.SubjectTree[T1], r *level[T2], subj []byte, cb func(subj []byte, entry *T1)) {
if r.numNodes() == 0 {
// For wildcards we can't avoid Match, but if it's a literal subject at
// this point, using Find is considerably cheaper.
if subjectHasWildcard(string(subj)) {
st.Match(subj, cb)
} else if e, ok := st.Find(subj); ok {
cb(subj, e)
}
return
}
nsubj := subj
if len(nsubj) > 0 {
nsubj = append(subj, '.')
}
switch {
case r.fwc != nil:
// We've reached a full wildcard, do a FWC match on the stree at this point
// and don't keep iterating downward.
nsubj := append(nsubj, '>')
st.Match(nsubj, cb)
case r.pwc != nil:
// We've found a partial wildcard. We'll keep iterating downwards, but first
// check whether there's interest at this level (without triggering dupes) and
// match if so.
nsubj := append(nsubj, '*')
if len(r.pwc.subs) > 0 && r.pwc.next != nil && r.pwc.next.numNodes() > 0 {
st.Match(nsubj, cb)
}
intersectStree(st, r.pwc.next, nsubj, cb)
case r.numNodes() > 0:
// Normal node with subject literals, keep iterating.
for t, n := range r.nodes {
nsubj := append(nsubj, t...)
intersectStree(st, n.next, nsubj, cb)
}
}
}
// Determine if a subject has any wildcard tokens.
func subjectHasWildcard(subject string) bool {
// This one exits earlier then !subjectIsLiteral(subject)
for i, c := range subject {
if c == pwc || c == fwc {
if (i == 0 || subject[i-1] == btsep) &&
(i+1 == len(subject) || subject[i+1] == btsep) {
return true
}
}
}
return false
}
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2021-2023 The NATS Authors
// Copyright 2021-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
+8 -9
View File
@@ -1,4 +1,4 @@
// Copyright 2019-2024 The NATS Authors
// Copyright 2019-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
@@ -1501,12 +1501,14 @@ func (a *Account) filteredStreams(filter string) []*stream {
var msets []*stream
for _, mset := range jsa.streams {
if filter != _EMPTY_ {
mset.cfgMu.RLock()
for _, subj := range mset.cfg.Subjects {
if SubjectsCollide(filter, subj) {
msets = append(msets, mset)
break
}
}
mset.cfgMu.RUnlock()
} else {
msets = append(msets, mset)
}
@@ -2147,14 +2149,11 @@ func (jsa *jsAccount) selectLimits(replicas int) (JetStreamAccountLimits, string
}
// Lock should be held.
func (jsa *jsAccount) countStreams(tier string, cfg *StreamConfig) int {
streams := len(jsa.streams)
if tier != _EMPTY_ {
streams = 0
for _, sa := range jsa.streams {
if isSameTier(&sa.cfg, cfg) {
streams++
}
func (jsa *jsAccount) countStreams(tier string, cfg *StreamConfig) (streams int) {
for _, sa := range jsa.streams {
// Don't count the stream toward the limit if it already exists.
if (tier == _EMPTY_ || isSameTier(&sa.cfg, cfg)) && sa.cfg.Name != cfg.Name {
streams++
}
}
return streams
+12 -11
View File
@@ -1,4 +1,4 @@
// Copyright 2020-2023 The NATS Authors
// Copyright 2020-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
@@ -765,7 +765,7 @@ func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, sub
s, rr := js.srv, js.apiSubs.Match(subject)
hdr, msg := c.msgParts(rmsg)
if len(getHeader(ClientInfoHdr, hdr)) == 0 {
if len(sliceHeader(ClientInfoHdr, hdr)) == 0 {
// Check if this is the system account. We will let these through for the account info only.
sacc := s.SystemAccount()
if sacc != acc {
@@ -1008,7 +1008,7 @@ func (s *Server) getRequestInfo(c *client, raw []byte) (pci *ClientInfo, acc *Ac
var ci ClientInfo
if len(hdr) > 0 {
if err := json.Unmarshal(getHeader(ClientInfoHdr, hdr), &ci); err != nil {
if err := json.Unmarshal(sliceHeader(ClientInfoHdr, hdr), &ci); err != nil {
return nil, nil, nil, nil, err
}
}
@@ -1873,13 +1873,14 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
if cc.meta != nil {
ourID = cc.meta.ID()
}
// We have seen cases where rg or rg.node is nil at this point,
// so check explicitly on those conditions and bail if that is
// the case.
bail := rg == nil || rg.node == nil || !rg.isMember(ourID)
// We have seen cases where rg is nil at this point,
// so check explicitly and bail if that is the case.
bail := rg == nil || !rg.isMember(ourID)
if !bail {
// We know we are a member here, if this group is new and we are preferred allow us to answer.
bail = rg.Preferred != ourID || time.Since(rg.node.Created()) > lostQuorumIntervalDefault
// Also, we have seen cases where rg.node is nil at this point,
// so check explicitly and bail if that is the case.
bail = rg.Preferred != ourID || (rg.node != nil && time.Since(rg.node.Created()) > lostQuorumIntervalDefault)
}
js.mu.RUnlock()
if bail {
@@ -4271,7 +4272,7 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
// Since these could wait on the Raft group lock, don't do so under the JS lock.
ourID := meta.ID()
groupLeader := meta.GroupLeader()
groupLeaderless := meta.Leaderless()
groupCreated := meta.Created()
js.mu.RLock()
@@ -4289,7 +4290,7 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
// Also capture if we think there is no meta leader.
var isLeaderLess bool
if !isLeader {
isLeaderLess = groupLeader == _EMPTY_ && time.Since(groupCreated) > lostQuorumIntervalDefault
isLeaderLess = groupLeaderless && time.Since(groupCreated) > lostQuorumIntervalDefault
}
js.mu.RUnlock()
@@ -4376,7 +4377,7 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
return
}
// If we are a member and we have a group leader or we had a previous leader consider bailing out.
if node.GroupLeader() != _EMPTY_ || node.HadPreviousLeader() {
if !node.Leaderless() || node.HadPreviousLeader() {
if leaderNotPartOfGroup {
resp.Error = NewJSConsumerOfflineError()
s.sendDelayedAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp), nil)
+108 -87
View File
@@ -1,4 +1,4 @@
// Copyright 2020-2024 The NATS Authors
// Copyright 2020-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
@@ -142,6 +142,7 @@ type streamAssignment struct {
responded bool
recovering bool
reassigning bool // i.e. due to placement issues, lack of resources, etc.
resetting bool // i.e. there was an error, and we're stopping and starting the stream
err error
}
@@ -444,108 +445,113 @@ func (cc *jetStreamCluster) isStreamCurrent(account, stream string) bool {
// isStreamHealthy will determine if the stream is up to date or very close.
// For R1 it will make sure the stream is present on this server.
func (js *jetStream) isStreamHealthy(acc *Account, sa *streamAssignment) bool {
func (js *jetStream) isStreamHealthy(acc *Account, sa *streamAssignment) error {
js.mu.RLock()
s, cc := js.srv, js.cluster
if cc == nil {
// Non-clustered mode
js.mu.RUnlock()
return true
return nil
}
// Pull the group out.
rg := sa.Group
if rg == nil {
if sa == nil || sa.Group == nil {
js.mu.RUnlock()
return false
return errors.New("stream assignment or group missing")
}
streamName := sa.Config.Name
node := rg.node
node := sa.Group.node
js.mu.RUnlock()
// First lookup stream and make sure its there.
mset, err := acc.lookupStream(streamName)
if err != nil {
return false
return errors.New("stream not found")
}
// If R1 we are good.
if node == nil {
return true
}
switch {
case mset.cfg.Replicas <= 1:
return nil // No further checks for R=1 streams
// Here we are a replicated stream.
// First make sure our monitor routine is running.
if !mset.isMonitorRunning() {
return false
}
case node == nil:
return errors.New("group node missing")
if node.Healthy() {
// Check if we are processing a snapshot and are catching up.
if !mset.isCatchingUp() {
return true
}
} else { // node != nil
if node != mset.raftNode() {
s.Warnf("Detected stream cluster node skew '%s > %s'", acc.GetName(), streamName)
node.Delete()
mset.resetClusteredState(nil)
}
case node != mset.raftNode():
s.Warnf("Detected stream cluster node skew '%s > %s'", acc.GetName(), streamName)
node.Delete()
mset.resetClusteredState(nil)
return errors.New("cluster node skew detected")
case !mset.isMonitorRunning():
return errors.New("monitor goroutine not running")
case !node.Healthy():
return errors.New("group node unhealthy")
case mset.isCatchingUp():
return errors.New("stream catching up")
default:
return nil
}
return false
}
// isConsumerHealthy will determine if the consumer is up to date.
// For R1 it will make sure the consunmer is present on this server.
func (js *jetStream) isConsumerHealthy(mset *stream, consumer string, ca *consumerAssignment) bool {
func (js *jetStream) isConsumerHealthy(mset *stream, consumer string, ca *consumerAssignment) error {
if mset == nil {
return false
return errors.New("stream missing")
}
js.mu.RLock()
cc := js.cluster
s, cc := js.srv, js.cluster
if cc == nil {
// Non-clustered mode
js.mu.RUnlock()
return true
return nil
}
// These are required.
if ca == nil || ca.Group == nil {
js.mu.RUnlock()
return false
return errors.New("consumer assignment or group missing")
}
s := js.srv
// Capture RAFT node from assignment.
node := ca.Group.node
js.mu.RUnlock()
// Check if not running at all.
o := mset.lookupConsumer(consumer)
if o == nil {
return false
return errors.New("consumer not found")
}
// Check RAFT node state.
if node == nil || node.Healthy() {
return true
} else if node != nil {
if node != o.raftNode() {
mset.mu.RLock()
accName, streamName := mset.acc.GetName(), mset.cfg.Name
mset.mu.RUnlock()
s.Warnf("Detected consumer cluster node skew '%s > %s > %s'", accName, streamName, consumer)
node.Delete()
o.deleteWithoutAdvisory()
rc, _ := o.replica()
switch {
case rc <= 1:
return nil // No further checks for R=1 consumers
// When we try to restart we nil out the node and reprocess the consumer assignment.
js.mu.Lock()
ca.Group.node = nil
js.mu.Unlock()
js.processConsumerAssignment(ca)
}
case node == nil:
return errors.New("group node missing")
case node != o.raftNode():
mset.mu.RLock()
accName, streamName := mset.acc.GetName(), mset.cfg.Name
mset.mu.RUnlock()
s.Warnf("Detected consumer cluster node skew '%s > %s > %s'", accName, streamName, consumer)
node.Delete()
o.deleteWithoutAdvisory()
// When we try to restart we nil out the node and reprocess the consumer assignment.
js.mu.Lock()
ca.Group.node = nil
js.mu.Unlock()
js.processConsumerAssignment(ca)
return errors.New("cluster node skew detected")
case !o.isMonitorRunning():
return errors.New("monitor goroutine not running")
case !node.Healthy():
return errors.New("group node unhealthy")
default:
return nil
}
return false
}
// subjectsOverlap checks all existing stream assignments for the account cross-cluster for subject overlap
@@ -819,7 +825,7 @@ func (js *jetStream) isLeaderless() bool {
// If we don't have a leader.
// Make sure we have been running for enough time.
if meta.GroupLeader() == _EMPTY_ && time.Since(meta.Created()) > lostQuorumIntervalDefault {
if meta.Leaderless() && time.Since(meta.Created()) > lostQuorumIntervalDefault {
return true
}
return false
@@ -851,7 +857,7 @@ func (js *jetStream) isGroupLeaderless(rg *raftGroup) bool {
node := rg.node
js.mu.RUnlock()
// If we don't have a leader.
if node.GroupLeader() == _EMPTY_ {
if node.Leaderless() {
// Threshold for jetstream startup.
const startupThreshold = 10 * time.Second
@@ -1067,7 +1073,7 @@ func (js *jetStream) checkForOrphans() {
// We only want to cleanup any orphans if we know we are current with the meta-leader.
meta := cc.meta
if meta == nil || meta.GroupLeader() == _EMPTY_ {
if meta == nil || meta.Leaderless() {
js.mu.Unlock()
s.Debugf("JetStream cluster skipping check for orphans, no meta-leader")
return
@@ -1366,7 +1372,7 @@ func (js *jetStream) monitorCluster() {
// If we have a current leader or had one in the past we can cancel this here since the metaleader
// will be in charge of all peer state changes.
// For cold boot only.
if n.GroupLeader() != _EMPTY_ || n.HadPreviousLeader() {
if !n.Leaderless() || n.HadPreviousLeader() {
lt.Stop()
continue
}
@@ -1581,10 +1587,11 @@ func (js *jetStream) applyMetaSnapshot(buf []byte, ru *recoveryUpdates, isRecove
}
if osa := js.streamAssignment(sa.Client.serviceAccount(), sa.Config.Name); osa != nil {
for _, ca := range osa.consumers {
if sa.consumers[ca.Name] == nil {
// Consumer was either removed, or recreated with a different raft group.
if nca := sa.consumers[ca.Name]; nca == nil {
caDel = append(caDel, ca)
} else if nca.Group != nil && ca.Group != nil && nca.Group.Name != ca.Group.Name {
caDel = append(caDel, ca)
} else {
caAdd = append(caAdd, ca)
}
}
}
@@ -2503,7 +2510,8 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
ce.ReturnToPool()
} else {
// Our stream was closed out from underneath of us, simply return here.
if err == errStreamClosed {
if err == errStreamClosed || err == errCatchupStreamStopped || err == ErrServerNotRunning {
aq.recycle(&ces)
return
}
s.Warnf("Error applying entries to '%s > %s': %v", accName, sa.Config.Name, err)
@@ -2549,7 +2557,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
// Always cancel if this was running.
stopDirectMonitoring()
} else if n.GroupLeader() != noLeader {
} else if !n.Leaderless() {
js.setStreamAssignmentRecovering(sa)
}
@@ -2913,6 +2921,9 @@ func (mset *stream) resetClusteredState(err error) bool {
}
s.Warnf("Resetting stream cluster state for '%s > %s'", sa.Client.serviceAccount(), sa.Config.Name)
// Mark stream assignment as resetting, so we don't double-account reserved resources.
// But only if we're not also releasing the resources as part of the delete.
sa.resetting = !shouldDelete
// Now wipe groups from assignments.
sa.Group.node = nil
var consumers []*consumerAssignment
@@ -3146,7 +3157,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
}
} else if e.Type == EntrySnapshot {
if mset == nil {
return nil
continue
}
// Everything operates on new replicated state. Will convert legacy snapshots to this for processing.
@@ -3216,7 +3227,6 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
mset.stop(true, false)
}
}
return nil
}
}
return nil
@@ -4046,7 +4056,7 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
js.mu.RLock()
s := js.srv
node := sa.Group.node
hadLeader := node == nil || node.GroupLeader() != noLeader
hadLeader := node == nil || !node.Leaderless()
offline := s.allPeersOffline(sa.Group)
var isMetaLeader bool
if cc := js.cluster; cc != nil {
@@ -5034,7 +5044,6 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
o.stopWithFlags(true, false, false, false)
}
}
return nil
} else if e.Type == EntryAddPeer {
// Ignore for now.
} else {
@@ -5994,15 +6003,16 @@ func groupName(prefix string, peers []string, storage StorageType) string {
return fmt.Sprintf("%s-R%d%s-%s", prefix, len(peers), storage.String()[:1], gns)
}
// returns stream count for this tier as well as applicable reservation size (not including reservations for cfg)
// returns stream count for this tier as well as applicable reservation size (not including cfg)
// jetStream read lock should be held
func tieredStreamAndReservationCount(asa map[string]*streamAssignment, tier string, cfg *StreamConfig) (int, int64) {
var numStreams int
var reservation int64
for _, sa := range asa {
if tier == _EMPTY_ || isSameTier(sa.Config, cfg) {
// Don't count the stream toward the limit if it already exists.
if (tier == _EMPTY_ || isSameTier(sa.Config, cfg)) && sa.Config.Name != cfg.Name {
numStreams++
if sa.Config.MaxBytes > 0 && sa.Config.Storage == cfg.Storage && sa.Config.Name != cfg.Name {
if sa.Config.MaxBytes > 0 && sa.Config.Storage == cfg.Storage {
// If tier is empty, all storage is flat and we should adjust for replicas.
// Otherwise if tiered, storage replication already taken into consideration.
if tier == _EMPTY_ && cfg.Replicas > 1 {
@@ -6084,7 +6094,14 @@ func (js *jetStream) jsClusteredStreamLimitsCheck(acc *Account, cfg *StreamConfi
numStreams, reservations := tieredStreamAndReservationCount(asa, tier, cfg)
// Check for inflight proposals...
if cc := js.cluster; cc != nil && cc.inflight != nil {
numStreams += len(cc.inflight[acc.Name])
streams := cc.inflight[acc.Name]
numStreams += len(streams)
// If inflight contains the same stream, don't count toward exceeding maximum.
if cfg != nil {
if _, ok := streams[cfg.Name]; ok {
numStreams--
}
}
}
if selectedLimits.MaxStreams > 0 && numStreams >= selectedLimits.MaxStreams {
return NewJSMaximumStreamsLimitError()
@@ -6192,7 +6209,7 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
// On success, add this as an inflight proposal so we can apply limits
// on concurrent create requests while this stream assignment has
// possibly not been processed yet.
if streams, ok := cc.inflight[acc.Name]; ok {
if streams, ok := cc.inflight[acc.Name]; ok && self == nil {
streams[cfg.Name] = &inflightInfo{rg, syncSubject}
}
}
@@ -7328,13 +7345,11 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
// Don't count DIRECTS.
total := 0
for cn, ca := range sa.consumers {
if action == ActionCreateOrUpdate {
// If the consumer name is specified and we think it already exists, then
// we're likely updating an existing consumer, so don't count it. Otherwise
// we will incorrectly return NewJSMaximumConsumersLimitError for an update.
if oname != _EMPTY_ && cn == oname && sa.consumers[oname] != nil {
continue
}
// If the consumer name is specified and we think it already exists, then
// we're likely updating an existing consumer, so don't count it. Otherwise
// we will incorrectly return NewJSMaximumConsumersLimitError for an update.
if oname != _EMPTY_ && cn == oname && sa.consumers[oname] != nil {
continue
}
if ca.Config != nil && !ca.Config.Direct {
total++
@@ -8035,6 +8050,12 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
err := node.Propose(esm)
if err == nil {
mset.clseq++
// If we are using the system account for NRG, add in the extra sent msgs and bytes to our account
// so that the end user / account owner has visibility.
if node.IsSystemAccount() && mset.acc != nil && r > 1 {
atomic.AddInt64(&mset.acc.outMsgs, int64(r-1))
atomic.AddInt64(&mset.acc.outBytes, int64(len(esm)*(r-1)))
}
}
// Check to see if we are being overrun.
@@ -8316,13 +8337,13 @@ RETRY:
// the semaphore.
releaseSyncOutSem()
if n.GroupLeader() == _EMPTY_ {
if n.Leaderless() {
// Prevent us from spinning if we've installed a snapshot from a leader but there's no leader online.
// We wait a bit to check if a leader has come online in the meantime, if so we can continue.
var canContinue bool
if numRetries == 0 {
time.Sleep(startInterval)
canContinue = n.GroupLeader() != _EMPTY_
canContinue = !n.Leaderless()
}
if !canContinue {
return fmt.Errorf("%w for stream '%s > %s'", errCatchupAbortedNoLeader, mset.account(), mset.name())
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2020-2021 The NATS Authors
// Copyright 2020-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
+19 -6
View File
@@ -1,4 +1,4 @@
// Copyright 2019-2024 The NATS Authors
// Copyright 2019-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
@@ -855,9 +855,18 @@ func (c *client) sendLeafConnect(clusterName string, headers bool) error {
pkey, _ := kp.PublicKey()
cinfo.Nkey = pkey
cinfo.Sig = sig
} else if userInfo := c.leaf.remote.curURL.User; userInfo != nil {
}
// In addition, and this is to allow auth callout, set user/password or
// token if applicable.
if userInfo := c.leaf.remote.curURL.User; userInfo != nil {
// For backward compatibility, if only username is provided, set both
// Token and User, not just Token.
cinfo.User = userInfo.Username()
cinfo.Pass, _ = userInfo.Password()
var ok bool
cinfo.Pass, ok = userInfo.Password()
if !ok {
cinfo.Token = cinfo.User
}
} else if c.leaf.remote.username != _EMPTY_ {
cinfo.User = c.leaf.remote.username
cinfo.Pass = c.leaf.remote.password
@@ -988,6 +997,7 @@ func (s *Server) createLeafNode(conn net.Conn, rURL *url.URL, remote *leafNodeCf
c.Noticef("Leafnode connection created%s %s", remoteSuffix, c.opts.Name)
var tlsFirst bool
var infoTimeout time.Duration
if remote != nil {
solicited = true
remote.Lock()
@@ -997,6 +1007,7 @@ func (s *Server) createLeafNode(conn net.Conn, rURL *url.URL, remote *leafNodeCf
c.leaf.isSpoke = true
}
tlsFirst = remote.TLSHandshakeFirst
infoTimeout = remote.FirstInfoTimeout
remote.Unlock()
c.acc = acc
} else {
@@ -1054,7 +1065,7 @@ func (s *Server) createLeafNode(conn net.Conn, rURL *url.URL, remote *leafNodeCf
}
}
// We need to wait for the info, but not for too long.
c.nc.SetReadDeadline(time.Now().Add(DEFAULT_LEAFNODE_INFO_WAIT))
c.nc.SetReadDeadline(time.Now().Add(infoTimeout))
}
// We will process the INFO from the readloop and finish by
@@ -1725,6 +1736,7 @@ type leafConnectInfo struct {
Sig string `json:"sig,omitempty"`
User string `json:"user,omitempty"`
Pass string `json:"pass,omitempty"`
Token string `json:"auth_token,omitempty"`
ID string `json:"server_id,omitempty"`
Domain string `json:"domain,omitempty"`
Name string `json:"name,omitempty"`
@@ -2771,7 +2783,7 @@ func (c *client) processInboundLeafMsg(msg []byte) {
// Now deal with gateways
if c.srv.gateway.enabled {
c.sendMsgToGateways(acc, msg, c.pa.subject, c.pa.reply, qnames)
c.sendMsgToGateways(acc, msg, c.pa.subject, c.pa.reply, qnames, true)
}
}
@@ -2887,6 +2899,7 @@ func (c *client) leafNodeSolicitWSConnection(opts *Options, rURL *url.URL, remot
compress := remote.Websocket.Compression
// By default the server will mask outbound frames, but it can be disabled with this option.
noMasking := remote.Websocket.NoMasking
infoTimeout := remote.FirstInfoTimeout
remote.RUnlock()
// Will do the client-side TLS handshake if needed.
tlsRequired, err := c.leafClientHandshakeIfNeeded(remote, opts)
@@ -2939,6 +2952,7 @@ func (c *client) leafNodeSolicitWSConnection(opts *Options, rURL *url.URL, remot
if noMasking {
req.Header.Add(wsNoMaskingHeader, wsNoMaskingValue)
}
c.nc.SetDeadline(time.Now().Add(infoTimeout))
if err := req.Write(c.nc); err != nil {
return nil, WriteError, err
}
@@ -2946,7 +2960,6 @@ func (c *client) leafNodeSolicitWSConnection(opts *Options, rURL *url.URL, remot
var resp *http.Response
br := bufio.NewReaderSize(c.nc, MAX_CONTROL_LINE_SIZE)
c.nc.SetReadDeadline(time.Now().Add(DEFAULT_LEAFNODE_INFO_WAIT))
resp, err = http.ReadResponse(br, req)
if err == nil &&
(resp.StatusCode != 101 ||
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2020 The NATS Authors
// Copyright 2012-2024 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
+38 -4
View File
@@ -1,4 +1,4 @@
// Copyright 2019-2024 The NATS Authors
// Copyright 2019-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
@@ -92,7 +92,7 @@ func (ms *memStore) UpdateConfig(cfg *StreamConfig) error {
// If the value is smaller, or was unset before, we need to enforce that.
if ms.maxp > 0 && (maxp == 0 || ms.maxp < maxp) {
lm := uint64(ms.maxp)
ms.fss.Iter(func(subj []byte, ss *SimpleState) bool {
ms.fss.IterFast(func(subj []byte, ss *SimpleState) bool {
if ss.Msgs > lm {
ms.enforcePerSubjectLimit(bytesToString(subj), ss)
}
@@ -196,6 +196,7 @@ func (ms *memStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts int
if ss != nil {
ss.Msgs++
ss.Last = seq
ss.lastNeedsUpdate = false
// Check per subject limits.
if ms.maxp > 0 && ss.Msgs > uint64(ms.maxp) {
ms.enforcePerSubjectLimit(subj, ss)
@@ -1012,6 +1013,8 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
ms.removeSeqPerSubject(sm.subj, seq)
// Must delete message after updating per-subject info, to be consistent with file store.
delete(ms.msgs, seq)
} else if !ms.dmap.IsEmpty() {
ms.dmap.Delete(seq)
}
}
if purged > ms.state.Msgs {
@@ -1032,9 +1035,10 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
ms.state.FirstSeq = seq
ms.state.FirstTime = time.Time{}
ms.state.LastSeq = seq - 1
// Reset msgs and fss.
// Reset msgs, fss and dmap.
ms.msgs = make(map[uint64]*StoreMsg)
ms.fss = stree.NewSubjectTree[SimpleState]()
ms.dmap.Empty()
}
ms.mu.Unlock()
@@ -1066,9 +1070,10 @@ func (ms *memStore) reset() error {
// Update msgs and bytes.
ms.state.Msgs = 0
ms.state.Bytes = 0
// Reset msgs and fss.
// Reset msgs, fss and dmap.
ms.msgs = make(map[uint64]*StoreMsg)
ms.fss = stree.NewSubjectTree[SimpleState]()
ms.dmap.Empty()
ms.mu.Unlock()
@@ -1102,6 +1107,8 @@ func (ms *memStore) Truncate(seq uint64) error {
ms.removeSeqPerSubject(sm.subj, i)
// Must delete message after updating per-subject info, to be consistent with file store.
delete(ms.msgs, i)
} else if !ms.dmap.IsEmpty() {
ms.dmap.Delete(i)
}
}
// Reset last.
@@ -1299,6 +1306,33 @@ func (ms *memStore) LoadNextMsg(filter string, wc bool, start uint64, smp *Store
return nil, ms.state.LastSeq, ErrStoreEOF
}
// Will load the next non-deleted msg starting at the start sequence and walking backwards.
func (ms *memStore) LoadPrevMsg(start uint64, smp *StoreMsg) (sm *StoreMsg, err error) {
ms.mu.RLock()
defer ms.mu.RUnlock()
if ms.msgs == nil {
return nil, ErrStoreClosed
}
if ms.state.Msgs == 0 || start < ms.state.FirstSeq {
return nil, ErrStoreEOF
}
if start > ms.state.LastSeq {
start = ms.state.LastSeq
}
for seq := start; seq >= ms.state.FirstSeq; seq-- {
if sm, ok := ms.msgs[seq]; ok {
if smp == nil {
smp = new(StoreMsg)
}
sm.copy(smp)
return smp, nil
}
}
return nil, ErrStoreEOF
}
// RemoveMsg will remove the message from this store.
// Will return the number of bytes removed.
func (ms *memStore) RemoveMsg(seq uint64) (bool, error) {
+110 -45
View File
@@ -1,4 +1,4 @@
// Copyright 2013-2024 The NATS Authors
// Copyright 2013-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
@@ -831,6 +831,7 @@ func (s *Server) Routez(routezOpts *RoutezOptions) (*Routez, error) {
OutBytes: r.outBytes,
NumSubs: uint32(len(r.subs)),
Import: r.opts.Import,
Pending: int(r.out.pb),
Export: r.opts.Export,
RTT: r.getRTT().String(),
Start: r.start,
@@ -1122,20 +1123,16 @@ func (s *Server) HandleStacksz(w http.ResponseWriter, r *http.Request) {
ResponseHandler(w, r, buf[:n])
}
type monitorIPQueue struct {
type IpqueueszStatusIPQ struct {
Pending int `json:"pending"`
InProgress int `json:"in_progress,omitempty"`
}
func (s *Server) HandleIPQueuesz(w http.ResponseWriter, r *http.Request) {
all, err := decodeBool(w, r, "all")
if err != nil {
return
}
qfilter := r.URL.Query().Get("queues")
queues := map[string]monitorIPQueue{}
type IpqueueszStatus map[string]IpqueueszStatusIPQ
func (s *Server) Ipqueuesz(opts *IpqueueszOptions) *IpqueueszStatus {
all, qfilter := opts.All, opts.Filter
queues := IpqueueszStatus{}
s.ipQueues.Range(func(k, v any) bool {
var pending, inProgress int
name := k.(string)
@@ -1152,9 +1149,23 @@ func (s *Server) HandleIPQueuesz(w http.ResponseWriter, r *http.Request) {
} else if qfilter != _EMPTY_ && !strings.Contains(name, qfilter) {
return true
}
queues[name] = monitorIPQueue{Pending: pending, InProgress: inProgress}
queues[name] = IpqueueszStatusIPQ{Pending: pending, InProgress: inProgress}
return true
})
return &queues
}
func (s *Server) HandleIPQueuesz(w http.ResponseWriter, r *http.Request) {
all, err := decodeBool(w, r, "all")
if err != nil {
return
}
qfilter := r.URL.Query().Get("queues")
queues := s.Ipqueuesz(&IpqueueszOptions{
All: all,
Filter: qfilter,
})
b, _ := json.MarshalIndent(queues, "", " ")
ResponseHandler(w, r, b)
@@ -1858,6 +1869,14 @@ type GatewayzOptions struct {
// AccountName will limit the list of accounts to that account name (makes Accounts implicit)
AccountName string `json:"account_name"`
// AccountSubscriptions indicates if subscriptions should be included in the results.
// Note: This is used only if `Accounts` or `AccountName` are specified.
AccountSubscriptions bool `json:"subscriptions"`
// AccountSubscriptionsDetail indicates if subscription details should be included in the results.
// Note: This is used only if `Accounts` or `AccountName` are specified.
AccountSubscriptionsDetail bool `json:"subscriptions_detail"`
}
// Gatewayz represents detailed information on Gateways
@@ -1880,12 +1899,14 @@ type RemoteGatewayz struct {
// AccountGatewayz represents interest mode for this account
type AccountGatewayz struct {
Name string `json:"name"`
InterestMode string `json:"interest_mode"`
NoInterestCount int `json:"no_interest_count,omitempty"`
InterestOnlyThreshold int `json:"interest_only_threshold,omitempty"`
TotalSubscriptions int `json:"num_subs,omitempty"`
NumQueueSubscriptions int `json:"num_queue_subs,omitempty"`
Name string `json:"name"`
InterestMode string `json:"interest_mode"`
NoInterestCount int `json:"no_interest_count,omitempty"`
InterestOnlyThreshold int `json:"interest_only_threshold,omitempty"`
TotalSubscriptions int `json:"num_subs,omitempty"`
NumQueueSubscriptions int `json:"num_queue_subs,omitempty"`
Subs []string `json:"subscriptions_list,omitempty"`
SubsDetail []SubDetail `json:"subscriptions_list_detail,omitempty"`
}
// Gatewayz returns a Gatewayz struct containing information about gateways.
@@ -2011,14 +2032,14 @@ func createOutboundAccountsGatewayz(opts *GatewayzOptions, gw *gateway) []*Accou
if !ok {
return nil
}
a := createAccountOutboundGatewayz(accName, ei)
a := createAccountOutboundGatewayz(opts, accName, ei)
return []*AccountGatewayz{a}
}
accs := make([]*AccountGatewayz, 0, 4)
gw.outsim.Range(func(k, v any) bool {
name := k.(string)
a := createAccountOutboundGatewayz(name, v)
a := createAccountOutboundGatewayz(opts, name, v)
accs = append(accs, a)
return true
})
@@ -2026,7 +2047,7 @@ func createOutboundAccountsGatewayz(opts *GatewayzOptions, gw *gateway) []*Accou
}
// Returns an AccountGatewayz for this gateway outbound connection
func createAccountOutboundGatewayz(name string, ei any) *AccountGatewayz {
func createAccountOutboundGatewayz(opts *GatewayzOptions, name string, ei any) *AccountGatewayz {
a := &AccountGatewayz{
Name: name,
InterestOnlyThreshold: gatewayMaxRUnsubBeforeSwitch,
@@ -2038,6 +2059,23 @@ func createAccountOutboundGatewayz(name string, ei any) *AccountGatewayz {
a.NoInterestCount = len(e.ni)
a.NumQueueSubscriptions = e.qsubs
a.TotalSubscriptions = int(e.sl.Count())
if opts.AccountSubscriptions || opts.AccountSubscriptionsDetail {
var subsa [4096]*subscription
subs := subsa[:0]
e.sl.All(&subs)
if opts.AccountSubscriptions {
a.Subs = make([]string, 0, len(subs))
} else {
a.SubsDetail = make([]SubDetail, 0, len(subs))
}
for _, sub := range subs {
if opts.AccountSubscriptions {
a.Subs = append(a.Subs, string(sub.subject))
} else {
a.SubsDetail = append(a.SubsDetail, newClientSubDetail(sub))
}
}
}
e.RUnlock()
} else {
a.InterestMode = Optimistic.String()
@@ -2129,6 +2167,10 @@ func (s *Server) HandleGatewayz(w http.ResponseWriter, r *http.Request) {
s.httpReqStats[GatewayzPath]++
s.mu.Unlock()
subs, subsDet, err := decodeSubs(w, r)
if err != nil {
return
}
accs, err := decodeBool(w, r, "accs")
if err != nil {
return
@@ -2140,9 +2182,11 @@ func (s *Server) HandleGatewayz(w http.ResponseWriter, r *http.Request) {
}
opts := &GatewayzOptions{
Name: gwName,
Accounts: accs,
AccountName: accName,
Name: gwName,
Accounts: accs,
AccountName: accName,
AccountSubscriptions: subs,
AccountSubscriptionsDetail: subsDet,
}
gw, err := s.Gatewayz(opts)
if err != nil {
@@ -2282,7 +2326,7 @@ type AccountStatz struct {
Accounts []*AccountStat `json:"account_statz"`
}
// LeafzOptions are options passed to Leafz
// AccountStatzOptions are options passed to account stats requests.
type AccountStatzOptions struct {
Accounts []string `json:"accounts"`
IncludeUnused bool `json:"include_unused"`
@@ -2760,6 +2804,18 @@ type ProfilezOptions struct {
Duration time.Duration `json:"duration,omitempty"`
}
// IpqueueszOptions are options passed to Ipqueuesz
type IpqueueszOptions struct {
All bool `json:"all"`
Filter string `json:"filter"`
}
// RaftzOptions are options passed to Raftz
type RaftzOptions struct {
AccountFilter string `json:"account"`
GroupFilter string `json:"group"`
}
// StreamDetail shows information about the stream state and its consumers.
type StreamDetail struct {
Name string `json:"name"`
@@ -3676,27 +3732,27 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
for stream, sa := range asa {
// Make sure we can look up
if !js.isStreamHealthy(acc, sa) {
if err := js.isStreamHealthy(acc, sa); err != nil {
if !details {
health.Status = na
health.Error = fmt.Sprintf("JetStream stream '%s > %s' is not current", accName, stream)
health.Error = fmt.Sprintf("JetStream stream '%s > %s' is not current: %s", accName, stream, err)
return health
}
health.Errors = append(health.Errors, HealthzError{
Type: HealthzErrorStream,
Account: accName,
Stream: stream,
Error: fmt.Sprintf("JetStream stream '%s > %s' is not current", accName, stream),
Error: fmt.Sprintf("JetStream stream '%s > %s' is not current: %s", accName, stream, err),
})
continue
}
mset, _ := acc.lookupStream(stream)
// Now check consumers.
for consumer, ca := range sa.consumers {
if !js.isConsumerHealthy(mset, consumer, ca) {
if err := js.isConsumerHealthy(mset, consumer, ca); err != nil {
if !details {
health.Status = na
health.Error = fmt.Sprintf("JetStream consumer '%s > %s > %s' is not current", acc, stream, consumer)
health.Error = fmt.Sprintf("JetStream consumer '%s > %s > %s' is not current: %s", acc, stream, consumer, err)
return health
}
health.Errors = append(health.Errors, HealthzError{
@@ -3704,7 +3760,7 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
Account: accName,
Stream: stream,
Consumer: consumer,
Error: fmt.Sprintf("JetStream consumer '%s > %s > %s' is not current", acc, stream, consumer),
Error: fmt.Sprintf("JetStream consumer '%s > %s > %s' is not current: %s", acc, stream, consumer, err),
})
}
}
@@ -3813,6 +3869,8 @@ type RaftzGroupPeer struct {
LastSeen string `json:"last_seen,omitempty"`
}
type RaftzStatus map[string]map[string]RaftzGroup
func (s *Server) HandleRaftz(w http.ResponseWriter, r *http.Request) {
if s.raftNodes == nil {
w.WriteHeader(404)
@@ -3820,20 +3878,34 @@ func (s *Server) HandleRaftz(w http.ResponseWriter, r *http.Request) {
return
}
gfilter := r.URL.Query().Get("group")
afilter := r.URL.Query().Get("acc")
groups := s.Raftz(&RaftzOptions{
AccountFilter: r.URL.Query().Get("acc"),
GroupFilter: r.URL.Query().Get("group"),
})
if groups == nil {
w.WriteHeader(404)
w.Write([]byte("No Raft nodes returned, check supplied filters"))
return
}
b, _ := json.MarshalIndent(groups, "", " ")
ResponseHandler(w, r, b)
}
func (s *Server) Raftz(opts *RaftzOptions) *RaftzStatus {
afilter, gfilter := opts.AccountFilter, opts.GroupFilter
if afilter == _EMPTY_ {
if sys := s.SystemAccount(); sys != nil {
afilter = sys.Name
} else {
w.WriteHeader(404)
w.Write([]byte("System account not found, the server may be shutting down"))
return
return nil
}
}
groups := map[string]RaftNode{}
infos := map[string]map[string]RaftzGroup{} // account -> group ID
infos := RaftzStatus{} // account -> group ID
s.rnMu.RLock()
if gfilter != _EMPTY_ {
@@ -3859,12 +3931,6 @@ func (s *Server) HandleRaftz(w http.ResponseWriter, r *http.Request) {
}
s.rnMu.RUnlock()
if len(groups) == 0 {
w.WriteHeader(404)
w.Write([]byte("No Raft nodes found, does the specified account/group exist?"))
return
}
for name, rg := range groups {
n, ok := rg.(*raft)
if n == nil || !ok {
@@ -3887,7 +3953,7 @@ func (s *Server) HandleRaftz(w http.ResponseWriter, r *http.Request) {
Applied: n.applied,
CatchingUp: n.catchup != nil,
Leader: n.leader,
EverHadLeader: n.pleader,
EverHadLeader: n.pleader.Load(),
Term: n.term,
Vote: n.vote,
PTerm: n.pterm,
@@ -3918,6 +3984,5 @@ func (s *Server) HandleRaftz(w http.ResponseWriter, r *http.Request) {
infos[n.accName][name] = info
}
b, _ := json.MarshalIndent(infos, "", " ")
ResponseHandler(w, r, b)
return &infos
}
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2013-2018 The NATS Authors
// Copyright 2013-2023 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2020-2023 The NATS Authors
// Copyright 2020-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2018 The NATS Authors
// Copyright 2018-2023 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2021-2023 The NATS Authors
// Copyright 2021-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 The NATS Authors
// Copyright 2023-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 The NATS Authors
// Copyright 2023-2024 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
+17 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2024 The NATS Authors
// Copyright 2012-2023 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
@@ -205,6 +205,11 @@ type RemoteLeafOpts struct {
DenyImports []string `json:"-"`
DenyExports []string `json:"-"`
// FirstInfoTimeout is the amount of time the server will wait for the
// initial INFO protocol from the remote server before closing the
// connection.
FirstInfoTimeout time.Duration `json:"-"`
// Compression options for this remote. Each remote could have a different
// setting and also be different from the LeafNode options.
Compression CompressionOpts `json:"-"`
@@ -290,6 +295,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"`
@@ -1570,6 +1576,10 @@ 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)
case "max_closed_clients":
o.MaxClosedClients = int(v.(int64))
default:
if au := atomic.LoadInt32(&allowUnknownTopLevelField); au == 0 && !tk.IsUsedVariable() {
err := &unknownConfigFieldErr{
@@ -2607,6 +2617,8 @@ func parseRemoteLeafNodes(v any, errors *[]error, warnings *[]error) ([]*RemoteL
*errors = append(*errors, err)
continue
}
case "first_info_timeout":
remote.FirstInfoTimeout = parseDuration(k, tk, v, errors, warnings)
default:
if !tk.IsUsedVariable() {
err := &unknownConfigFieldErr{
@@ -5193,6 +5205,10 @@ func setBaselineOptions(opts *Options) {
c.Mode = CompressionS2Auto
}
}
// Set default first info timeout value if not set.
if r.FirstInfoTimeout <= 0 {
r.FirstInfoTimeout = DEFAULT_LEAFNODE_INFO_WAIT
}
}
}
+17 -15
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2020 The NATS Authors
// Copyright 2012-2024 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
@@ -35,20 +35,21 @@ type parseState struct {
}
type pubArg struct {
arg []byte
pacache []byte
origin []byte
account []byte
subject []byte
deliver []byte
mapped []byte
reply []byte
szb []byte
hdb []byte
queues [][]byte
size int
hdr int
psi []*serviceImport
arg []byte
pacache []byte
origin []byte
account []byte
subject []byte
deliver []byte
mapped []byte
reply []byte
szb []byte
hdb []byte
queues [][]byte
size int
hdr int
psi []*serviceImport
delivered bool // Only used for service imports
}
// Parser constants
@@ -500,6 +501,7 @@ func (c *client) parse(buf []byte) error {
// Drop all pub args
c.pa.arg, c.pa.pacache, c.pa.origin, c.pa.account, c.pa.subject, c.pa.mapped = nil, nil, nil, nil, nil, nil
c.pa.reply, c.pa.hdr, c.pa.size, c.pa.szb, c.pa.hdb, c.pa.queues = nil, -1, 0, nil, nil, nil
c.pa.delivered = false
lmsg = false
case OP_A:
switch b {
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2015-2018 The NATS Authors
// Copyright 2015-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2015-2018 The NATS Authors
// Copyright 2015-2022 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2015-2018 The NATS Authors
// Copyright 2015-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2015-2018 The NATS Authors
// Copyright 2015-2024 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
+40 -17
View File
@@ -1,4 +1,4 @@
// Copyright 2020-2024 The NATS Authors
// Copyright 2020-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
@@ -52,6 +52,7 @@ type RaftNode interface {
Current() bool
Healthy() bool
Term() uint64
Leaderless() bool
GroupLeader() string
HadPreviousLeader() bool
StepDown(preferred ...string) error
@@ -77,6 +78,7 @@ type RaftNode interface {
Stop()
WaitForStop()
Delete()
IsSystemAccount() bool
}
type WAL interface {
@@ -174,9 +176,10 @@ type raft struct {
c *client // Internal client for subscriptions
js *jetStream // JetStream, if running, to see if we are out of resources
dflag bool // Debug flag
pleader bool // Has the group ever had a leader?
observer bool // The node is observing, i.e. not participating in voting
dflag bool // Debug flag
hasleader atomic.Bool // Is there a group leader right now?
pleader atomic.Bool // Has the group ever had a leader?
observer bool // The node is observing, i.e. not participating in voting
extSt extensionState // Extension state
@@ -542,6 +545,12 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
return n, nil
}
// Whether we are using the system account or not.
// In 2.10.x this is always true as there is no account NRG like in 2.11.x.
func (n *raft) IsSystemAccount() bool {
return true
}
// outOfResources checks to see if we are out of resources.
func (n *raft) outOfResources() bool {
js := n.js
@@ -830,7 +839,7 @@ func (n *raft) AdjustBootClusterSize(csz int) error {
n.Lock()
defer n.Unlock()
if n.leader != noLeader || n.pleader {
if n.leader != noLeader || n.pleader.Load() {
return errAdjustBootCluster
}
// Same floor as bootstrap.
@@ -1386,9 +1395,7 @@ func (n *raft) Healthy() bool {
// HadPreviousLeader indicates if this group ever had a leader.
func (n *raft) HadPreviousLeader() bool {
n.RLock()
defer n.RUnlock()
return n.pleader
return n.pleader.Load()
}
// GroupLeader returns the current leader of the group.
@@ -1401,6 +1408,17 @@ func (n *raft) GroupLeader() string {
return n.leader
}
// Leaderless is a lockless way of finding out if the group has a
// leader or not. Use instead of GroupLeader in hot paths.
func (n *raft) Leaderless() bool {
if n == nil {
return true
}
// Negated because we want the default state of hasLeader to be
// false until the first setLeader() call.
return !n.hasleader.Load()
}
// Guess the best next leader. Stepdown will check more thoroughly.
// Lock should be held.
func (n *raft) selectNextLeader() string {
@@ -3146,8 +3164,9 @@ func (n *raft) resetWAL() {
// Lock should be held
func (n *raft) updateLeader(newLeader string) {
n.leader = newLeader
if !n.pleader && newLeader != noLeader {
n.pleader = true
n.hasleader.Store(newLeader != _EMPTY_)
if !n.pleader.Load() && newLeader != noLeader {
n.pleader.Store(true)
}
}
@@ -3424,8 +3443,13 @@ CONTINUE:
if l > paeWarnThreshold && l%paeWarnModulo == 0 {
n.warn("%d append entries pending", len(n.pae))
}
} else if l%paeWarnModulo == 0 {
n.debug("Not saving to append entries pending")
} else {
// Invalidate cache entry at this index, we might have
// stored it previously with a different value.
delete(n.pae, n.pindex)
if l%paeWarnModulo == 0 {
n.debug("Not saving to append entries pending")
}
}
} else {
// This is a replay on startup so just take the appendEntry version.
@@ -4000,11 +4024,10 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
n.vote = vr.candidate
n.writeTermVote()
n.resetElectionTimeout()
} else {
if vr.term >= n.term && n.vote == noVote {
n.term = vr.term
n.resetElect(randCampaignTimeout())
}
} else if n.vote == noVote && n.State() != Candidate {
// We have a more up-to-date log, and haven't voted yet.
// Start campaigning earlier, but only if not candidate already, as that would short-circuit us.
n.resetElect(randCampaignTimeout())
}
// Term might have changed, make sure response has the most current
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2021-2022 The NATS Authors
// Copyright 2021-2021 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
+16 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2017-2023 The NATS Authors
// Copyright 2017-2024 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
@@ -916,6 +916,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()
@@ -1623,6 +1636,8 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
if new != old {
diffOpts = append(diffOpts, &profBlockRateReload{newValue: new})
}
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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2018 The NATS Authors
// Copyright 2018-2020 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2013-2023 The NATS Authors
// Copyright 2013-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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2020-2023 The NATS Authors
// Copyright 2020-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2018 The NATS Authors
// Copyright 2012-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2019 The NATS Authors
// Copyright 2012-2022 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2019 The NATS Authors
// Copyright 2012-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2019 The NATS Authors
// Copyright 2012-2023 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
+2 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2019-2024 The NATS Authors
// Copyright 2019-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
@@ -91,6 +91,7 @@ type StreamStore interface {
LoadNextMsg(filter string, wc bool, start uint64, smp *StoreMsg) (sm *StoreMsg, skip uint64, err error)
LoadNextMsgMulti(sl *Sublist, start uint64, smp *StoreMsg) (sm *StoreMsg, skip uint64, err error)
LoadLastMsg(subject string, sm *StoreMsg) (*StoreMsg, error)
LoadPrevMsg(start uint64, smp *StoreMsg) (sm *StoreMsg, err error)
RemoveMsg(seq uint64) (bool, error)
EraseMsg(seq uint64) (bool, error)
Purge() (uint64, error)
+67 -51
View File
@@ -1,4 +1,4 @@
// Copyright 2019-2024 The NATS Authors
// Copyright 2019-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
@@ -16,7 +16,6 @@ package server
import (
"archive/tar"
"bytes"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
@@ -33,6 +32,7 @@ import (
"time"
"github.com/klauspost/compress/s2"
"github.com/nats-io/nats-server/v2/server/gsl"
"github.com/nats-io/nuid"
)
@@ -271,10 +271,10 @@ type stream struct {
// For processing consumers without main stream lock.
clsMu sync.RWMutex
cList []*consumer // Consumer list.
sch chan struct{} // Channel to signal consumers.
sigq *ipQueue[*cMsg] // Intra-process queue for the messages to signal to the consumers.
csl *Sublist // Consumer subscription list.
cList []*consumer // Consumer list.
sch chan struct{} // Channel to signal consumers.
sigq *ipQueue[*cMsg] // Intra-process queue for the messages to signal to the consumers.
csl *gsl.GenericSublist[*consumer] // Consumer subscription list.
// For non limits policy streams when they process an ack before the actual msg.
// Can happen in stretch clusters, multi-cloud, or during catchup for a restarted server.
@@ -660,15 +660,25 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
}
// Set our stream assignment if in clustered mode.
reserveResources := true
if sa != nil {
mset.setStreamAssignment(sa)
// If the stream is resetting we must not double-account resources, they were already accounted for.
js.mu.Lock()
if sa.resetting {
reserveResources, sa.resetting = false, false
}
js.mu.Unlock()
}
// Setup our internal send go routine.
mset.setupSendCapabilities()
// Reserve resources if MaxBytes present.
mset.js.reserveStreamResources(&mset.cfg)
if reserveResources {
mset.js.reserveStreamResources(&mset.cfg)
}
// Call directly to set leader if not in clustered mode.
// This can be called though before we actually setup clustering, so check both.
@@ -3488,16 +3498,21 @@ func (mset *stream) setStartingSequenceForSources(iNames map[string]struct{}) {
}
var smv StoreMsg
for seq := state.LastSeq; seq >= state.FirstSeq; seq-- {
sm, err := mset.store.LoadMsg(seq, &smv)
if err != nil || len(sm.hdr) == 0 {
for seq := state.LastSeq; seq >= state.FirstSeq; {
sm, err := mset.store.LoadPrevMsg(seq, &smv)
if err == ErrStoreEOF || err != nil {
break
}
seq = sm.seq - 1
if len(sm.hdr) == 0 {
continue
}
ss := getHeader(JSStreamSource, sm.hdr)
if len(ss) == 0 {
continue
}
streamName, indexName, sseq := streamAndSeq(string(ss))
streamName, indexName, sseq := streamAndSeq(bytesToString(ss))
if _, ok := iNames[indexName]; ok {
si := mset.sources[indexName]
@@ -3603,9 +3618,13 @@ func (mset *stream) startingSequenceForSources() {
}
var smv StoreMsg
for seq := state.LastSeq; seq >= state.FirstSeq; seq-- {
sm, err := mset.store.LoadMsg(seq, &smv)
if err != nil || sm == nil || len(sm.hdr) == 0 {
for seq := state.LastSeq; ; {
sm, err := mset.store.LoadPrevMsg(seq, &smv)
if err == ErrStoreEOF || err != nil {
break
}
seq = sm.seq - 1
if len(sm.hdr) == 0 {
continue
}
ss := getHeader(JSStreamSource, sm.hdr)
@@ -3613,7 +3632,7 @@ func (mset *stream) startingSequenceForSources() {
continue
}
streamName, iName, sseq := streamAndSeq(string(ss))
streamName, iName, sseq := streamAndSeq(bytesToString(ss))
if iName == _EMPTY_ { // Pre-2.10 message header means it's a match for any source using that stream name
for _, ssi := range mset.cfg.Sources {
if streamName == ssi.Name || (ssi.External != nil && streamName == ssi.Name+":"+getHash(ssi.External.ApiPrefix)) {
@@ -3932,9 +3951,14 @@ func (mset *stream) storeUpdates(md, bd int64, seq uint64, subj string) {
if md == -1 && seq > 0 && subj != _EMPTY_ {
// We use our consumer list mutex here instead of the main stream lock since it may be held already.
mset.clsMu.RLock()
// TODO(dlc) - Do sublist like signaling so we do not have to match?
for _, o := range mset.cList {
o.decStreamPending(seq, subj)
if mset.csl != nil {
mset.csl.Match(subj, func(o *consumer) {
o.decStreamPending(seq, subj)
})
} else {
for _, o := range mset.cList {
o.decStreamPending(seq, subj)
}
}
mset.clsMu.RUnlock()
} else if md < 0 {
@@ -4806,24 +4830,14 @@ func (mset *stream) signalConsumersLoop() {
// This will update and signal all consumers that match.
func (mset *stream) signalConsumers(subj string, seq uint64) {
mset.clsMu.RLock()
if mset.csl == nil {
mset.clsMu.RUnlock()
defer mset.clsMu.RUnlock()
csl := mset.csl
if csl == nil {
return
}
r := mset.csl.Match(subj)
mset.clsMu.RUnlock()
if len(r.psubs) == 0 {
return
}
// Encode the sequence here.
var eseq [8]byte
var le = binary.LittleEndian
le.PutUint64(eseq[:], seq)
msg := eseq[:]
for _, sub := range r.psubs {
sub.icb(sub, nil, nil, subj, _EMPTY_, msg)
}
csl.Match(subj, func(o *consumer) {
o.processStreamSignal(seq)
})
}
// Internal message for use by jetstream subsystem.
@@ -5367,10 +5381,10 @@ func (mset *stream) setConsumer(o *consumer) {
mset.clsMu.Lock()
mset.cList = append(mset.cList, o)
if mset.csl == nil {
mset.csl = NewSublistWithCache()
mset.csl = gsl.NewSublist[*consumer]()
}
for _, sub := range o.signalSubs() {
mset.csl.Insert(sub)
mset.csl.Insert(sub, o)
}
mset.clsMu.Unlock()
}
@@ -5396,7 +5410,7 @@ func (mset *stream) removeConsumer(o *consumer) {
// Always remove from the leader sublist.
if mset.csl != nil {
for _, sub := range o.signalSubs() {
mset.csl.Remove(sub)
mset.csl.Remove(sub, o)
}
}
mset.clsMu.Unlock()
@@ -5418,7 +5432,7 @@ func (mset *stream) swapSigSubs(o *consumer, newFilters []string) {
if o.sigSubs != nil {
if mset.csl != nil {
for _, sub := range o.sigSubs {
mset.csl.Remove(sub)
mset.csl.Remove(sub, o)
}
}
o.sigSubs = nil
@@ -5426,19 +5440,17 @@ func (mset *stream) swapSigSubs(o *consumer, newFilters []string) {
if o.isLeader() {
if mset.csl == nil {
mset.csl = NewSublistWithCache()
mset.csl = gsl.NewSublist[*consumer]()
}
// If no filters are preset, add fwcs to sublist for that consumer.
if newFilters == nil {
sub := &subscription{subject: []byte(fwcs), icb: o.processStreamSignal}
mset.csl.Insert(sub)
o.sigSubs = append(o.sigSubs, sub)
mset.csl.Insert(fwcs, o)
o.sigSubs = append(o.sigSubs, fwcs)
// If there are filters, add their subjects to sublist.
} else {
for _, filter := range newFilters {
sub := &subscription{subject: []byte(filter), icb: o.processStreamSignal}
mset.csl.Insert(sub)
o.sigSubs = append(o.sigSubs, sub)
mset.csl.Insert(filter, o)
o.sigSubs = append(o.sigSubs, filter)
}
}
}
@@ -5671,16 +5683,17 @@ func (mset *stream) clearPreAck(o *consumer, seq uint64) {
}
// ackMsg is called into from a consumer when we have a WorkQueue or Interest Retention Policy.
func (mset *stream) ackMsg(o *consumer, seq uint64) {
// Returns whether the message at seq was removed as a result of the ACK.
func (mset *stream) ackMsg(o *consumer, seq uint64) bool {
if seq == 0 {
return
return false
}
// Don't make this RLock(). We need to have only 1 running at a time to gauge interest across all consumers.
mset.mu.Lock()
if mset.closed.Load() || mset.cfg.Retention == LimitsPolicy {
mset.mu.Unlock()
return
return false
}
store := mset.store
@@ -5691,7 +5704,9 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
if seq > state.LastSeq {
mset.registerPreAck(o, seq)
mset.mu.Unlock()
return
// We have not removed the message, but should still signal so we could retry later
// since we potentially need to remove it then.
return true
}
// Always clear pre-ack if here.
@@ -5700,7 +5715,7 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
// Make sure this sequence is not below our first sequence.
if seq < state.FirstSeq {
mset.mu.Unlock()
return
return false
}
var shouldRemove bool
@@ -5716,7 +5731,7 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
// If nothing else to do.
if !shouldRemove {
return
return false
}
// If we are here we should attempt to remove.
@@ -5724,6 +5739,7 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
// This should not happen, but being pedantic.
mset.registerPreAckLock(o, seq)
}
return true
}
// Snapshot creates a snapshot for the stream and possibly consumers.
+28 -6
View File
@@ -1,4 +1,4 @@
// Copyright 2023-2024 The NATS Authors
// Copyright 2023-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
@@ -124,13 +124,22 @@ func (t *SubjectTree[T]) Match(filter []byte, cb func(subject []byte, val *T)) {
t.match(t.root, parts, _pre[:0], cb)
}
// Iter will walk all entries in the SubjectTree lexographically. The callback can return false to terminate the walk.
func (t *SubjectTree[T]) Iter(cb func(subject []byte, val *T) bool) {
// IterOrdered will walk all entries in the SubjectTree lexographically. The callback can return false to terminate the walk.
func (t *SubjectTree[T]) IterOrdered(cb func(subject []byte, val *T) bool) {
if t == nil || t.root == nil {
return
}
var _pre [256]byte
t.iter(t.root, _pre[:0], cb)
t.iter(t.root, _pre[:0], true, cb)
}
// IterFast will walk all entries in the SubjectTree with no guarantees of ordering. The callback can return false to terminate the walk.
func (t *SubjectTree[T]) IterFast(cb func(subject []byte, val *T) bool) {
if t == nil || t.root == nil {
return
}
var _pre [256]byte
t.iter(t.root, _pre[:0], false, cb)
}
// Internal methods
@@ -369,7 +378,7 @@ func (t *SubjectTree[T]) match(n node, parts [][]byte, pre []byte, cb func(subje
}
// Interal iter function to walk nodes in lexigraphical order.
func (t *SubjectTree[T]) iter(n node, pre []byte, cb func(subject []byte, val *T) bool) bool {
func (t *SubjectTree[T]) iter(n node, pre []byte, ordered bool, cb func(subject []byte, val *T) bool) bool {
if n.isLeaf() {
ln := n.(*leaf[T])
return cb(append(pre, ln.suffix...), &ln.value)
@@ -378,6 +387,19 @@ func (t *SubjectTree[T]) iter(n node, pre []byte, cb func(subject []byte, val *T
bn := n.base()
// Note that this append may reallocate, but it doesn't modify "pre" at the "iter" callsite.
pre = append(pre, bn.prefix...)
// Not everything requires lexicographical sorting, so support a fast path for iterating in
// whatever order the stree has things stored instead.
if !ordered {
for _, cn := range n.children() {
if cn == nil {
continue
}
if !t.iter(cn, pre, false, cb) {
return false
}
}
return true
}
// Collect nodes since unsorted.
var _nodes [256]node
nodes := _nodes[:0]
@@ -390,7 +412,7 @@ func (t *SubjectTree[T]) iter(n node, pre []byte, cb func(subject []byte, val *T
slices.SortStableFunc(nodes, func(a, b node) int { return bytes.Compare(a.path(), b.path()) })
// Now walk the nodes in order and call into next iter.
for i := range nodes {
if !t.iter(nodes[i], pre, cb) {
if !t.iter(nodes[i], pre, true, cb) {
return false
}
}
+1 -9
View File
@@ -1,4 +1,4 @@
// Copyright 2023-2024 The NATS Authors
// Copyright 2023-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
@@ -55,11 +55,3 @@ func pivot[N position](subject []byte, pos N) byte {
}
return subject[pos]
}
// TODO(dlc) - Can be removed with Go 1.21 once server is on Go 1.22.
func min(a, b int) int {
if a < b {
return a
}
return b
}
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 The NATS Authors
// Copyright 2023-2024 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
+8 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2016-2024 The NATS Authors
// Copyright 2016-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
@@ -1744,7 +1744,13 @@ func IntersectStree[T any](st *stree.SubjectTree[T], sl *Sublist, cb func(subj [
func intersectStree[T any](st *stree.SubjectTree[T], r *level, subj []byte, cb func(subj []byte, entry *T)) {
if r.numNodes() == 0 {
st.Match(subj, cb)
// For wildcards we can't avoid Match, but if it's a literal subject at
// this point, using Find is considerably cheaper.
if subjectHasWildcard(bytesToString(subj)) {
st.Match(subj, cb)
} else if e, ok := st.Find(subj); ok {
cb(subj, e)
}
return
}
nsubj := subj
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2019 The NATS Authors
// Copyright 2019-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2019 The NATS Authors
// Copyright 2019-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2019 The NATS Authors
// Copyright 2019-2021 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2019 The NATS Authors
// Copyright 2019-2024 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2022 The NATS Authors
// Copyright 2022-2023 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
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2012-2019 The NATS Authors
// Copyright 2012-2024 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
+11 -8
View File
@@ -1,4 +1,4 @@
// Copyright 2020-2023 The NATS Authors
// Copyright 2020-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
@@ -1311,6 +1311,9 @@ func (c *client) wsCollapsePtoNB() (net.Buffers, int64) {
}
if usz <= wsCompressThreshold {
compress = false
if cp := c.ws.compressor; cp != nil {
cp.Reset(nil)
}
}
}
if compress && len(nb) > 0 {
@@ -1331,13 +1334,11 @@ func (c *client) wsCollapsePtoNB() (net.Buffers, int64) {
for len(b) > 0 {
n, err := cp.Write(b)
if err != nil {
if err == io.EOF {
break
}
c.Errorf("Error during compression: %v", err)
c.markConnAsClosed(WriteError)
nbPoolPut(b)
return nil, 0
// Whatever this error is, it'll be handled by the cp.Flush()
// call below, as the same error will be returned there.
// Let the outer loop return all the buffers back to the pool
// and fall through naturally.
break
}
b = b[n:]
}
@@ -1346,6 +1347,7 @@ func (c *client) wsCollapsePtoNB() (net.Buffers, int64) {
if err := cp.Flush(); err != nil {
c.Errorf("Error during compression: %v", err)
c.markConnAsClosed(WriteError)
cp.Reset(nil)
return nil, 0
}
b := buf.Bytes()
@@ -1461,6 +1463,7 @@ func (c *client) wsCollapsePtoNB() (net.Buffers, int64) {
bufs = append(bufs, c.ws.closeMsg)
c.ws.fs += int64(len(c.ws.closeMsg))
c.ws.closeMsg = nil
c.ws.compressor = nil
}
c.ws.frames = nil
return bufs, c.ws.fs