diff --git a/go.mod b/go.mod index c135c10ae..0143431b0 100644 --- a/go.mod +++ b/go.mod @@ -57,7 +57,7 @@ require ( github.com/mitchellh/mapstructure v1.5.0 github.com/mna/pigeon v1.2.1 github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 - github.com/nats-io/nats-server/v2 v2.10.2 + github.com/nats-io/nats-server/v2 v2.10.4 github.com/oklog/run v1.1.0 github.com/olekukonko/tablewriter v0.0.5 github.com/onsi/ginkgo v1.16.5 @@ -243,7 +243,7 @@ require ( github.com/json-iterator/go v1.1.12 // indirect github.com/juliangruber/go-intersect v1.1.0 // indirect github.com/kevinburke/ssh_config v1.2.0 // indirect - github.com/klauspost/compress v1.17.0 // indirect + github.com/klauspost/compress v1.17.2 // indirect github.com/klauspost/cpuid/v2 v2.1.0 // indirect github.com/leodido/go-urn v1.2.0 // indirect github.com/libregraph/oidc-go v1.0.0 // indirect @@ -272,8 +272,8 @@ require ( github.com/modern-go/reflect2 v1.0.2 // indirect github.com/mschoch/smat v0.2.0 // indirect github.com/nats-io/jwt/v2 v2.5.2 // indirect - github.com/nats-io/nats.go v1.30.2 // indirect - github.com/nats-io/nkeys v0.4.5 // indirect + github.com/nats-io/nats.go v1.31.0 // indirect + github.com/nats-io/nkeys v0.4.6 // indirect github.com/nats-io/nuid v1.0.1 // indirect github.com/nxadm/tail v1.4.8 // indirect github.com/opencontainers/runtime-spec v1.0.3-0.20210326190908-1c3f411f0417 // indirect diff --git a/go.sum b/go.sum index eec949df9..3e3f08942 100644 --- a/go.sum +++ b/go.sum @@ -1568,8 +1568,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE= github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= -github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM= -github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= +github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4= +github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= @@ -1723,12 +1723,12 @@ github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRW github.com/namedotcom/go v0.0.0-20180403034216-08470befbe04/go.mod h1:5sN+Lt1CaY4wsPvgQH/jsuJi4XO2ssZbdsIizr4CVC8= github.com/nats-io/jwt/v2 v2.5.2 h1:DhGH+nKt+wIkDxM6qnVSKjokq5t59AZV5HRcFW0zJwU= github.com/nats-io/jwt/v2 v2.5.2/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI= -github.com/nats-io/nats-server/v2 v2.10.2 h1:2o/OOyc/dxeMCQtrF1V/9er0SU0A3LKhDlv/+rqreBM= -github.com/nats-io/nats-server/v2 v2.10.2/go.mod h1:lzrskZ/4gyMAh+/66cCd+q74c6v7muBypzfWhP/MAaM= -github.com/nats-io/nats.go v1.30.2 h1:aloM0TGpPorZKQhbAkdCzYDj+ZmsJDyeo3Gkbr72NuY= -github.com/nats-io/nats.go v1.30.2/go.mod h1:dcfhUgmQNN4GJEfIb2f9R7Fow+gzBF4emzDHrVBd5qM= -github.com/nats-io/nkeys v0.4.5 h1:Zdz2BUlFm4fJlierwvGK+yl20IAKUm7eV6AAZXEhkPk= -github.com/nats-io/nkeys v0.4.5/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64= +github.com/nats-io/nats-server/v2 v2.10.4 h1:uB9xcwon3tPXWAdmTJqqqC6cie3yuPWHJjjTBgaPNus= +github.com/nats-io/nats-server/v2 v2.10.4/go.mod h1:eWm2JmHP9Lqm2oemB6/XGi0/GwsZwtWf8HIPUsh+9ns= +github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E= +github.com/nats-io/nats.go v1.31.0/go.mod h1:di3Bm5MLsoB4Bx61CBTsxuarI36WbhAwOm8QrW39+i8= +github.com/nats-io/nkeys v0.4.6 h1:IzVe95ru2CT6ta874rt9saQRkWfe2nFj1NtvYSLqMzY= +github.com/nats-io/nkeys v0.4.6/go.mod h1:4DxZNzenSVd1cYQoAa8948QY3QDjrHfcfVADymtkpts= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/nbio/st v0.0.0-20140626010706-e9e8d9816f32/go.mod h1:9wM+0iRr9ahx58uYLpLIr5fm8diHn0JbqRycJi6w0Ms= diff --git a/vendor/github.com/klauspost/compress/flate/inflate.go b/vendor/github.com/klauspost/compress/flate/inflate.go index 414c0bea9..2f410d64f 100644 --- a/vendor/github.com/klauspost/compress/flate/inflate.go +++ b/vendor/github.com/klauspost/compress/flate/inflate.go @@ -120,8 +120,9 @@ func (h *huffmanDecoder) init(lengths []int) bool { const sanity = false if h.chunks == nil { - h.chunks = &[huffmanNumChunks]uint16{} + h.chunks = new([huffmanNumChunks]uint16) } + if h.maxRead != 0 { *h = huffmanDecoder{chunks: h.chunks, links: h.links} } @@ -175,6 +176,7 @@ func (h *huffmanDecoder) init(lengths []int) bool { } h.maxRead = min + chunks := h.chunks[:] for i := range chunks { chunks[i] = 0 @@ -202,8 +204,7 @@ func (h *huffmanDecoder) init(lengths []int) bool { if cap(h.links[off]) < numLinks { h.links[off] = make([]uint16, numLinks) } else { - links := h.links[off][:0] - h.links[off] = links[:numLinks] + h.links[off] = h.links[off][:numLinks] } } } else { @@ -277,7 +278,7 @@ func (h *huffmanDecoder) init(lengths []int) bool { return true } -// The actual read interface needed by NewReader. +// Reader is the actual read interface needed by NewReader. // If the passed in io.Reader does not also have ReadByte, // the NewReader will introduce its own buffering. type Reader interface { @@ -285,6 +286,18 @@ type Reader interface { io.ByteReader } +type step uint8 + +const ( + copyData step = iota + 1 + nextBlock + huffmanBytesBuffer + huffmanBytesReader + huffmanBufioReader + huffmanStringsReader + huffmanGenericReader +) + // Decompress state. type decompressor struct { // Input source. @@ -303,7 +316,7 @@ type decompressor struct { // Next step in the decompression, // and decompression state. - step func(*decompressor) + step step stepState int err error toRead []byte @@ -342,7 +355,7 @@ func (f *decompressor) nextBlock() { // compressed, fixed Huffman tables f.hl = &fixedHuffmanDecoder f.hd = nil - f.huffmanBlockDecoder()() + f.huffmanBlockDecoder() if debugDecode { fmt.Println("predefinied huffman block") } @@ -353,7 +366,7 @@ func (f *decompressor) nextBlock() { } f.hl = &f.h1 f.hd = &f.h2 - f.huffmanBlockDecoder()() + f.huffmanBlockDecoder() if debugDecode { fmt.Println("dynamic huffman block") } @@ -379,14 +392,16 @@ func (f *decompressor) Read(b []byte) (int, error) { if f.err != nil { return 0, f.err } - f.step(f) + + f.doStep() + if f.err != nil && len(f.toRead) == 0 { f.toRead = f.dict.readFlush() // Flush what's left in case of error } } } -// Support the io.WriteTo interface for io.Copy and friends. +// WriteTo implements the io.WriteTo interface for io.Copy and friends. func (f *decompressor) WriteTo(w io.Writer) (int64, error) { total := int64(0) flushed := false @@ -410,7 +425,7 @@ func (f *decompressor) WriteTo(w io.Writer) (int64, error) { return total, f.err } if f.err == nil { - f.step(f) + f.doStep() } if len(f.toRead) == 0 && f.err != nil && !flushed { f.toRead = f.dict.readFlush() // Flush what's left in case of error @@ -631,7 +646,7 @@ func (f *decompressor) copyData() { if f.dict.availWrite() == 0 || f.copyLen > 0 { f.toRead = f.dict.readFlush() - f.step = (*decompressor).copyData + f.step = copyData return } f.finishBlock() @@ -644,7 +659,28 @@ func (f *decompressor) finishBlock() { } f.err = io.EOF } - f.step = (*decompressor).nextBlock + f.step = nextBlock +} + +func (f *decompressor) doStep() { + switch f.step { + case copyData: + f.copyData() + case nextBlock: + f.nextBlock() + case huffmanBytesBuffer: + f.huffmanBytesBuffer() + case huffmanBytesReader: + f.huffmanBytesReader() + case huffmanBufioReader: + f.huffmanBufioReader() + case huffmanStringsReader: + f.huffmanStringsReader() + case huffmanGenericReader: + f.huffmanGenericReader() + default: + panic("BUG: unexpected step state") + } } // noEOF returns err, unless err == io.EOF, in which case it returns io.ErrUnexpectedEOF. @@ -747,7 +783,7 @@ func (f *decompressor) Reset(r io.Reader, dict []byte) error { h1: f.h1, h2: f.h2, dict: f.dict, - step: (*decompressor).nextBlock, + step: nextBlock, } f.dict.init(maxMatchOffset, dict) return nil @@ -768,7 +804,7 @@ func NewReader(r io.Reader) io.ReadCloser { f.r = makeReader(r) f.bits = new([maxNumLit + maxNumDist]int) f.codebits = new([numCodes]int) - f.step = (*decompressor).nextBlock + f.step = nextBlock f.dict.init(maxMatchOffset, nil) return &f } @@ -787,7 +823,7 @@ func NewReaderDict(r io.Reader, dict []byte) io.ReadCloser { f.r = makeReader(r) f.bits = new([maxNumLit + maxNumDist]int) f.codebits = new([numCodes]int) - f.step = (*decompressor).nextBlock + f.step = nextBlock f.dict.init(maxMatchOffset, dict) return &f } diff --git a/vendor/github.com/klauspost/compress/flate/inflate_gen.go b/vendor/github.com/klauspost/compress/flate/inflate_gen.go index 61342b6b8..2b2f993f7 100644 --- a/vendor/github.com/klauspost/compress/flate/inflate_gen.go +++ b/vendor/github.com/klauspost/compress/flate/inflate_gen.go @@ -85,7 +85,7 @@ readLiteral: dict.writeByte(byte(v)) if dict.availWrite() == 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanBytesBuffer + f.step = huffmanBytesBuffer f.stepState = stateInit f.b, f.nb = fb, fnb return @@ -251,7 +251,7 @@ copyHistory: if dict.availWrite() == 0 || f.copyLen > 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanBytesBuffer // We need to continue this work + f.step = huffmanBytesBuffer // We need to continue this work f.stepState = stateDict f.b, f.nb = fb, fnb return @@ -336,7 +336,7 @@ readLiteral: dict.writeByte(byte(v)) if dict.availWrite() == 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanBytesReader + f.step = huffmanBytesReader f.stepState = stateInit f.b, f.nb = fb, fnb return @@ -502,7 +502,7 @@ copyHistory: if dict.availWrite() == 0 || f.copyLen > 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanBytesReader // We need to continue this work + f.step = huffmanBytesReader // We need to continue this work f.stepState = stateDict f.b, f.nb = fb, fnb return @@ -587,7 +587,7 @@ readLiteral: dict.writeByte(byte(v)) if dict.availWrite() == 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanBufioReader + f.step = huffmanBufioReader f.stepState = stateInit f.b, f.nb = fb, fnb return @@ -753,7 +753,7 @@ copyHistory: if dict.availWrite() == 0 || f.copyLen > 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanBufioReader // We need to continue this work + f.step = huffmanBufioReader // We need to continue this work f.stepState = stateDict f.b, f.nb = fb, fnb return @@ -838,7 +838,7 @@ readLiteral: dict.writeByte(byte(v)) if dict.availWrite() == 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanStringsReader + f.step = huffmanStringsReader f.stepState = stateInit f.b, f.nb = fb, fnb return @@ -1004,7 +1004,7 @@ copyHistory: if dict.availWrite() == 0 || f.copyLen > 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanStringsReader // We need to continue this work + f.step = huffmanStringsReader // We need to continue this work f.stepState = stateDict f.b, f.nb = fb, fnb return @@ -1089,7 +1089,7 @@ readLiteral: dict.writeByte(byte(v)) if dict.availWrite() == 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanGenericReader + f.step = huffmanGenericReader f.stepState = stateInit f.b, f.nb = fb, fnb return @@ -1255,7 +1255,7 @@ copyHistory: if dict.availWrite() == 0 || f.copyLen > 0 { f.toRead = dict.readFlush() - f.step = (*decompressor).huffmanGenericReader // We need to continue this work + f.step = huffmanGenericReader // We need to continue this work f.stepState = stateDict f.b, f.nb = fb, fnb return @@ -1265,19 +1265,19 @@ copyHistory: // Not reached } -func (f *decompressor) huffmanBlockDecoder() func() { +func (f *decompressor) huffmanBlockDecoder() { switch f.r.(type) { case *bytes.Buffer: - return f.huffmanBytesBuffer + f.huffmanBytesBuffer() case *bytes.Reader: - return f.huffmanBytesReader + f.huffmanBytesReader() case *bufio.Reader: - return f.huffmanBufioReader + f.huffmanBufioReader() case *strings.Reader: - return f.huffmanStringsReader + f.huffmanStringsReader() case Reader: - return f.huffmanGenericReader + f.huffmanGenericReader() default: - return f.huffmanGenericReader + f.huffmanGenericReader() } } diff --git a/vendor/github.com/klauspost/compress/s2/encode.go b/vendor/github.com/klauspost/compress/s2/encode.go index e6c231021..0c9088adf 100644 --- a/vendor/github.com/klauspost/compress/s2/encode.go +++ b/vendor/github.com/klauspost/compress/s2/encode.go @@ -57,7 +57,7 @@ func Encode(dst, src []byte) []byte { // The function returns -1 if no improvement could be achieved. // Using actual compression will most often produce better compression than the estimate. func EstimateBlockSize(src []byte) (d int) { - if len(src) < 6 || int64(len(src)) > 0xffffffff { + if len(src) <= inputMargin || int64(len(src)) > 0xffffffff { return -1 } if len(src) <= 1024 { diff --git a/vendor/github.com/klauspost/compress/s2/encode_best.go b/vendor/github.com/klauspost/compress/s2/encode_best.go index 1d13e869a..47bac7423 100644 --- a/vendor/github.com/klauspost/compress/s2/encode_best.go +++ b/vendor/github.com/klauspost/compress/s2/encode_best.go @@ -157,6 +157,9 @@ func encodeBlockBest(dst, src []byte, dict *Dict) (d int) { return m } matchDict := func(candidate, s int, first uint32, rep bool) match { + if s >= MaxDictSrcOffset { + return match{offset: candidate, s: s} + } // Calculate offset as if in continuous array with s offset := -len(dict.dict) + candidate if best.length != 0 && best.s-best.offset == s-offset && !rep { diff --git a/vendor/github.com/klauspost/compress/s2/encode_go.go b/vendor/github.com/klauspost/compress/s2/encode_go.go index 0d39c7b0e..6b393c34d 100644 --- a/vendor/github.com/klauspost/compress/s2/encode_go.go +++ b/vendor/github.com/klauspost/compress/s2/encode_go.go @@ -316,6 +316,7 @@ func matchLen(a []byte, b []byte) int { return len(a) + checked } +// input must be > inputMargin func calcBlockSize(src []byte) (d int) { // Initialize the hash table. const ( @@ -501,6 +502,7 @@ emitRemainder: return d } +// length must be > inputMargin. func calcBlockSizeSmall(src []byte) (d int) { // Initialize the hash table. const ( diff --git a/vendor/github.com/nats-io/nats-server/v2/server/client.go b/vendor/github.com/nats-io/nats-server/v2/server/client.go index e3364c8a8..6aba4395a 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/client.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/client.go @@ -141,6 +141,7 @@ const ( expectConnect // Marks if this connection is expected to send a CONNECT connectProcessFinished // Marks if this connection has finished the connect process. compressionNegotiated // Marks if this connection has negotiated compression level with remote. + didTLSFirst // Marks if this connection requested and was accepted doing the TLS handshake first (prior to INFO). ) // set the flag (would be equivalent to set the boolean to true) @@ -2800,6 +2801,11 @@ func (c *client) addShadowSubscriptions(acc *Account, sub *subscription) error { ) acc.mu.RLock() + // If this is from a service import, ignore. + if sub.si { + acc.mu.RUnlock() + return nil + } subj := string(sub.subject) if len(acc.imports.streams) > 0 { tokens = tokenizeSubjectIntoSlice(tsa[:0], subj) @@ -5226,7 +5232,7 @@ func (c *client) reconnect() { // Check for a solicited route. If it was, start up a reconnect unless // we are already connected to the other end. - if c.isSolicitedRoute() || retryImplicit { + if didSolicit := c.isSolicitedRoute(); didSolicit || retryImplicit { srv.mu.Lock() defer srv.mu.Unlock() @@ -5236,7 +5242,7 @@ func (c *client) reconnect() { rtype := c.route.routeType rurl := c.route.url accName := string(c.route.accName) - checkRID := accName == _EMPTY_ && srv.routesPoolSize <= 1 && rid != _EMPTY_ + checkRID := accName == _EMPTY_ && srv.getOpts().Cluster.PoolSize < 1 && rid != _EMPTY_ c.mu.Unlock() // It is possible that the server is being shutdown. @@ -5246,6 +5252,14 @@ func (c *client) reconnect() { } if checkRID && srv.routes[rid] != nil { + // This is the case of "no pool". Make sure that the registered one + // is upgraded to solicited if the connection trying to reconnect + // was a solicited one. + if didSolicit { + if remote := srv.routes[rid][0]; remote != nil { + upgradeRouteToSolicited(remote, rurl, rtype) + } + } srv.Debugf("Not attempting reconnect for solicited route, already connected to %q", rid) return } else if rid == srv.info.ID { @@ -5305,10 +5319,10 @@ func (c *client) getAccAndResultFromCache() (*Account, *SublistResult) { // Check our cache. if pac, ok = c.in.pacache[string(c.pa.pacache)]; ok { // Check the genid to see if it's still valid. - // sl could be swapped out on reload so need to lock. - pac.acc.mu.RLock() + // Since v2.10.0, the config reload of accounts has been fixed + // and an account's sublist pointer should not change, so no need to + // lock to access it. sl := pac.acc.sl - pac.acc.mu.RUnlock() if genid := atomic.LoadUint64(&sl.genid); genid != pac.genid { ok = false @@ -5320,15 +5334,15 @@ func (c *client) getAccAndResultFromCache() (*Account, *SublistResult) { } if !ok { - // Match correct account and sublist. - if acc, _ = c.srv.LookupAccount(string(c.pa.account)); acc == nil { - return nil, nil + if c.kind == ROUTER && len(c.route.accName) > 0 { + acc = c.acc + } else { + // Match correct account and sublist. + if acc, _ = c.srv.LookupAccount(string(c.pa.account)); acc == nil { + return nil, nil + } } - - // sl could be swapped out on reload so need to lock. - acc.mu.RLock() sl := acc.sl - acc.mu.RUnlock() // Match against the account sublist. r = sl.Match(string(c.pa.subject)) diff --git a/vendor/github.com/nats-io/nats-server/v2/server/const.go b/vendor/github.com/nats-io/nats-server/v2/server/const.go index 64ec6b626..603a3ae44 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/const.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/const.go @@ -41,7 +41,7 @@ var ( const ( // VERSION is the current version for the server. - VERSION = "2.10.2" + VERSION = "2.10.4" // PROTO is the currently supported protocol. // 0 was the original @@ -82,6 +82,12 @@ const ( // TLS_TIMEOUT is the TLS wait time. TLS_TIMEOUT = 2 * time.Second + // DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY is the default amount of + // time for the server to wait for the TLS handshake with a client to + // be initiated before falling back to sending the INFO protocol first. + // See TLSHandshakeFirst and TLSHandshakeFirstFallback options. + DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY = 50 * time.Millisecond + // AUTH_TIMEOUT is the authorization wait time. AUTH_TIMEOUT = 2 * time.Second diff --git a/vendor/github.com/nats-io/nats-server/v2/server/consumer.go b/vendor/github.com/nats-io/nats-server/v2/server/consumer.go index 15fbd1286..1f96e772e 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/consumer.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/consumer.go @@ -272,6 +272,10 @@ var ( AckNext = []byte("+NXT") // Terminate delivery of the message. AckTerm = []byte("+TERM") + + // reasons to supply when terminating messages using limits + ackTermLimitsReason = "Message deleted by stream limits" + ackTermUnackedLimitsReason = "Unacknowledged message was deleted" ) // Calculate accurate replicas for the consumer config with the parent stream config. @@ -292,6 +296,7 @@ type consumer struct { // This will be checked in checkPending to abort processing // and let ack be processed in priority. awl int64 + leader atomic.Bool mu sync.RWMutex js *jetStream mset *stream @@ -767,6 +772,13 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri if action == ActionCreate && !reflect.DeepEqual(*config, eo.config()) { return nil, NewJSConsumerAlreadyExistsError() } + // Check for overlapping subjects. + if mset.cfg.Retention == WorkQueuePolicy { + subjects := gatherSubjectFilters(config.FilterSubject, config.FilterSubjects) + if !mset.partitionUnique(cName, subjects) { + return nil, NewJSConsumerWQConsumerNotUniqueError() + } + } err := eo.updateConfig(config) if err == nil { return eo, nil @@ -805,7 +817,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri if len(subjects) == 0 { mset.mu.Unlock() return nil, NewJSConsumerWQMultipleUnfilteredError() - } else if !mset.partitionUnique(subjects) { + } else if !mset.partitionUnique(cName, subjects) { // Prior to v2.9.7, on a stream with WorkQueue policy, the servers // were not catching the error of having multiple consumers with // overlapping filter subjects depending on the scope, for instance @@ -1124,19 +1136,21 @@ func (o *consumer) isLeader() bool { func (o *consumer) setLeader(isLeader bool) { o.mu.RLock() - mset := o.mset - isRunning := o.ackSub != nil + mset, closed := o.mset, o.closed + movingToClustered := o.node != nil && o.pch == nil + wasLeader := o.leader.Swap(isLeader) o.mu.RUnlock() // If we are here we have a change in leader status. if isLeader { - if mset == nil { + if closed || mset == nil { return } - if isRunning { + + if wasLeader { // If we detect we are scaling up, make sure to create clustered routines and channels. - o.mu.Lock() - if o.node != nil && o.pch == nil { + if movingToClustered { + o.mu.Lock() // We are moving from R1 to clustered. o.pch = make(chan struct{}, 1) go o.loopAndForwardProposals(o.qch) @@ -1146,8 +1160,8 @@ func (o *consumer) setLeader(isLeader bool) { default: } } + o.mu.Unlock() } - o.mu.Unlock() return } @@ -1981,8 +1995,12 @@ func (o *consumer) processAck(subject, reply string, hdr int, rmsg []byte) { o.processNak(sseq, dseq, dc, msg) case bytes.Equal(msg, AckProgress): o.progressUpdate(sseq) - case bytes.Equal(msg, AckTerm): - o.processTerm(sseq, dseq, dc) + case bytes.HasPrefix(msg, AckTerm): + var reason string + if buf := msg[len(AckTerm):]; len(buf) > 0 { + reason = string(bytes.TrimSpace(buf)) + } + o.processTerm(sseq, dseq, dc, reason) } // Ack the ack if requested. @@ -2309,7 +2327,7 @@ func (o *consumer) processNak(sseq, dseq, dc uint64, nak []byte) { } // Process a TERM -func (o *consumer) processTerm(sseq, dseq, dc uint64) { +func (o *consumer) processTerm(sseq, dseq, dc uint64, reason string) { // Treat like an ack to suppress redelivery. o.processAckMsg(sseq, dseq, dc, false) @@ -2328,6 +2346,7 @@ func (o *consumer) processTerm(sseq, dseq, dc uint64) { ConsumerSeq: dseq, StreamSeq: sseq, Deliveries: dc, + Reason: reason, Domain: o.srv.getOpts().JetStreamDomain, } @@ -2536,7 +2555,11 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo { // If we are replicated and we are not the leader we need to pull certain data from our store. if rg != nil && rg.node != nil && !o.isLeader() && o.store != nil { - state, _ := o.store.BorrowState() + state, err := o.store.BorrowState() + if err != nil { + o.mu.Unlock() + return nil + } info.Delivered.Consumer, info.Delivered.Stream = state.Delivered.Consumer, state.Delivered.Stream info.AckFloor.Consumer, info.AckFloor.Stream = state.AckFloor.Consumer, state.AckFloor.Stream info.NumAckPending = len(state.Pending) @@ -3636,7 +3659,7 @@ func (o *consumer) checkAckFloor() { o.mu.RUnlock() // If it was pending for us, get rid of it. if isPending { - o.processTerm(seq, p.Sequence, rdc) + o.processTerm(seq, p.Sequence, rdc, ackTermLimitsReason) } } } else if numPending > 0 { @@ -3661,7 +3684,7 @@ func (o *consumer) checkAckFloor() { for i := 0; i < len(toTerm); i += 3 { seq, dseq, rdc := toTerm[i], toTerm[i+1], toTerm[i+2] - o.processTerm(seq, dseq, rdc) + o.processTerm(seq, dseq, rdc, ackTermLimitsReason) } } @@ -4634,16 +4657,6 @@ type lastSeqSkipList struct { seqs []uint64 } -// Will create a skip list for us from a store's subjects state. -func createLastSeqSkipList(mss map[string]SimpleState) []uint64 { - seqs := make([]uint64, 0, len(mss)) - for _, ss := range mss { - seqs = append(seqs, ss.Last) - } - sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] }) - return seqs -} - // Let's us know we have a skip list, which is for deliver last per subject and we are just starting. // Lock should be held. func (o *consumer) hasSkipListPending() bool { @@ -4674,37 +4687,49 @@ func (o *consumer) selectStartingSeqNo() { } } } else if o.cfg.DeliverPolicy == DeliverLastPerSubject { - if o.subjf == nil { - if mss := o.mset.store.SubjectsState(o.cfg.FilterSubject); len(mss) > 0 { - o.lss = &lastSeqSkipList{ - resume: state.LastSeq, - seqs: createLastSeqSkipList(mss), - } - o.sseq = o.lss.seqs[0] + // If our parent stream is set to max msgs per subject of 1 this is just + // a normal consumer at this point. We can avoid any heavy lifting. + if o.mset.cfg.MaxMsgsPer == 1 { + o.sseq = state.FirstSeq + } else { + // A threshold for when we switch from get last msg to subjects state. + const numSubjectsThresh = 256 + lss := &lastSeqSkipList{resume: state.LastSeq} + var filters []string + if o.subjf == nil { + filters = append(filters, o.cfg.FilterSubject) } else { - // If no mapping info just set to last. + for _, filter := range o.subjf { + filters = append(filters, filter.subject) + } + } + for _, filter := range filters { + if st := o.mset.store.SubjectsTotals(filter); len(st) < numSubjectsThresh { + var smv StoreMsg + for subj := range st { + if sm, err := o.mset.store.LoadLastMsg(subj, &smv); err == nil { + lss.seqs = append(lss.seqs, sm.seq) + } + } + } else if mss := o.mset.store.SubjectsState(filter); len(mss) > 0 { + for _, ss := range mss { + lss.seqs = append(lss.seqs, ss.Last) + } + } + } + // Sort the skip list if needed. + if len(lss.seqs) > 1 { + sort.Slice(lss.seqs, func(i, j int) bool { + return lss.seqs[j] > lss.seqs[i] + }) + } + if len(lss.seqs) == 0 { o.sseq = state.LastSeq + } else { + o.sseq = lss.seqs[0] } - return - } - lss := &lastSeqSkipList{ - resume: state.LastSeq, - } - for _, filter := range o.subjf { - if mss := o.mset.store.SubjectsState(filter.subject); len(mss) > 0 { - lss.seqs = append(lss.seqs, createLastSeqSkipList(mss)...) - } - } - if len(lss.seqs) == 0 { - o.sseq = state.LastSeq - } - // Sort the skip list - sort.Slice(lss.seqs, func(i, j int) bool { - return lss.seqs[j] > lss.seqs[i] - }) - o.lss = lss - if len(o.lss.seqs) != 0 { - o.sseq = o.lss.seqs[0] + // Assign skip list. + o.lss = lss } } else if o.cfg.OptStartTime != nil { // If we are here we are time based. @@ -5137,11 +5162,10 @@ func (o *consumer) decStreamPending(sseq uint64, subj string) { o.mu.Unlock() // If it was pending process it like an ack. - // TODO(dlc) - we could do a term here instead with a reason to generate the advisory. if wasPending { // We could have lock for stream so do this in a go routine. // TODO(dlc) - We should do this with ipq vs naked go routines. - go o.processTerm(sseq, p.Sequence, rdc) + go o.processTerm(sseq, p.Sequence, rdc, ackTermUnackedLimitsReason) } } diff --git a/vendor/github.com/nats-io/nats-server/v2/server/events.go b/vendor/github.com/nats-io/nats-server/v2/server/events.go index 0f761a47c..601ed85a0 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/events.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/events.go @@ -1598,14 +1598,16 @@ func (s *Server) shutdownEventing() { s.mu.Lock() clearTimer(&s.sys.sweeper) clearTimer(&s.sys.stmr) - sys := s.sys + rc := s.sys.resetCh + s.sys.resetCh = nil + wg := &s.sys.wg s.mu.Unlock() // We will queue up a shutdown event and wait for the // internal send loop to exit. s.sendShutdownEvent() - sys.wg.Wait() - close(sys.resetCh) + wg.Wait() + close(rc) s.mu.Lock() defer s.mu.Unlock() diff --git a/vendor/github.com/nats-io/nats-server/v2/server/filestore.go b/vendor/github.com/nats-io/nats-server/v2/server/filestore.go index 78e448111..3157e6784 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/filestore.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/filestore.go @@ -176,6 +176,7 @@ type fileStore struct { blks []*msgBlock bim map[uint32]*msgBlock psim map[string]*psi + tsl int hh hash.Hash64 qch chan struct{} fch chan struct{} @@ -428,7 +429,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim prior := fs.state // Reset anything that could have been set from above. fs.state = StreamState{} - fs.psim = make(map[string]*psi) + fs.psim, fs.tsl = make(map[string]*psi), 0 fs.bim = make(map[uint32]*msgBlock) fs.blks = nil fs.tombs = nil @@ -467,7 +468,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim // Check if we have any left over tombstones to process. if len(fs.tombs) > 0 { for _, seq := range fs.tombs { - fs.removeMsg(seq, false, false, false) + fs.removeMsg(seq, false, true, false) fs.removeFromLostData(seq) } // Not needed after this phase. @@ -486,7 +487,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim // If we have max msgs per subject make sure the is also enforced. if fs.cfg.MaxMsgsPer > 0 { - fs.enforceMsgPerSubjectLimit() + fs.enforceMsgPerSubjectLimit(false) } // Grab first sequence for check below while we have lock. @@ -589,7 +590,7 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error { } if fs.cfg.MaxMsgsPer > 0 && fs.cfg.MaxMsgsPer < old_cfg.MaxMsgsPer { - fs.enforceMsgPerSubjectLimit() + fs.enforceMsgPerSubjectLimit(true) } fs.mu.Unlock() @@ -937,6 +938,16 @@ func (mb *msgBlock) ensureLastChecksumLoaded() { copy(mb.lchk[0:], mb.lastChecksum()) } +// Perform a recover but do not update PSIM. +// Lock should be held. +func (fs *fileStore) recoverMsgBlockNoSubjectUpdates(index uint32) (*msgBlock, error) { + psim := fs.psim + fs.psim = nil + mb, err := fs.recoverMsgBlock(index) + fs.psim = psim + return mb, err +} + // Lock held on entry func (fs *fileStore) recoverMsgBlock(index uint32) (*msgBlock, error) { mb := fs.initMsgBlock(index) @@ -1218,6 +1229,8 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, []uint64, error) { mb.clearCacheAndOffset() buf, err := mb.loadBlock(nil) + defer recycleMsgBlockBuf(buf) + if err != nil || len(buf) == 0 { var ld *LostStreamData // No data to rebuild from here. @@ -1553,7 +1566,7 @@ func (fs *fileStore) recoverFullState() (rerr error) { // Check for per subject info. if numSubjects := int(readU64()); numSubjects > 0 { - fs.psim = make(map[string]*psi, numSubjects) + fs.psim, fs.tsl = make(map[string]*psi, numSubjects), 0 for i := 0; i < numSubjects; i++ { if lsubj := int(readU64()); lsubj > 0 { if bi+lsubj > len(buf) { @@ -1570,6 +1583,7 @@ func (fs *fileStore) recoverFullState() (rerr error) { psi.lblk = psi.fblk } fs.psim[subj] = psi + fs.tsl += len(subj) } } } @@ -1646,18 +1660,18 @@ func (fs *fileStore) recoverFullState() (rerr error) { return errPriorState } if matched = bytes.Equal(mb.lastChecksum(), lchk[:]); !matched { - // Remove the last message block since we will re-process below. + // Remove the last message block since recover will add in the new one. fs.removeMsgBlockFromList(mb) + if nmb, err := fs.recoverMsgBlockNoSubjectUpdates(mb.index); err != nil && !os.IsNotExist(err) { + os.Remove(fn) + return errCorruptState + } else if nmb != nil { + fs.adjustAccounting(mb, nmb) + } } // We may need to check other blocks. Even if we matched last checksum we will see if there is another block. - // If we did not match we re-process the last block. - start := blkIndex - if matched { - start++ - } - - for bi := start; ; bi++ { + for bi := blkIndex + 1; ; bi++ { nmb, err := fs.recoverMsgBlock(bi) if err != nil { if os.IsNotExist(err) { @@ -1668,13 +1682,6 @@ func (fs *fileStore) recoverFullState() (rerr error) { return err } if nmb != nil { - // Check if we have to account for a partial message block. - if !matched && mb != nil && mb.index == nmb.index { - if err := fs.adjustAccounting(mb, nmb); err != nil { - fs.warn("Stream state could not adjust accounting") - return err - } - } // Update top level accounting. if fs.state.FirstSeq == 0 || nmb.first.seq < fs.state.FirstSeq { fs.state.FirstSeq = nmb.first.seq @@ -1691,9 +1698,9 @@ func (fs *fileStore) recoverFullState() (rerr error) { } // adjustAccounting will be called when a stream state was only partially accounted for -// with a message block, e.g. additional records were added after the stream state. +// within a message block, e.g. additional records were added after the stream state. // Lock should be held. -func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) error { +func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) { nmb.mu.Lock() defer nmb.mu.Unlock() @@ -1703,23 +1710,29 @@ func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) error { } nmb.ensurePerSubjectInfoLoaded() - // Walk all the original mb's sequences that were included in the stream state. + // Walk only new messages and update accounting at fs level. Any messages that should have + // triggered limits exceeded will be handled after the recovery and prior to the stream + // being available to the system. var smv StoreMsg - for seq := mb.first.seq; seq <= mb.last.seq; seq++ { - // If we had already declared it deleted we can move on since you can not undelete. - if mb.dmap.Exists(seq) { - continue - } - // Lookup the message. + for seq := mb.last.seq + 1; seq <= nmb.last.seq; seq++ { + // Lookup the message. If an error will be deleted, so can skip. sm, err := nmb.cacheLookup(seq, &smv) if err != nil { - return err + continue } // Since we found it we just need to adjust fs totals and psim. - fs.state.Msgs-- - fs.state.Bytes -= fileStoreMsgSize(sm.subj, sm.hdr, sm.msg) + fs.state.Msgs++ + fs.state.Bytes += fileStoreMsgSize(sm.subj, sm.hdr, sm.msg) if len(sm.subj) > 0 && fs.psim != nil { - fs.removePerSubject(sm.subj) + if info, ok := fs.psim[sm.subj]; ok { + info.total++ + if nmb.index > info.lblk { + info.lblk = nmb.index + } + } else { + fs.psim[sm.subj] = &psi{total: 1, fblk: nmb.index, lblk: nmb.index} + fs.tsl += len(sm.subj) + } } } @@ -1729,7 +1742,15 @@ func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) error { nmb.first = mb.first } - return nil + // Update top level accounting. + if fs.state.FirstSeq == 0 || nmb.first.seq < fs.state.FirstSeq { + fs.state.FirstSeq = nmb.first.seq + fs.state.FirstTime = time.Unix(0, nmb.first.ts).UTC() + } + if nmb.last.seq > fs.state.LastSeq { + fs.state.LastSeq = nmb.last.seq + fs.state.LastTime = time.Unix(0, nmb.last.ts).UTC() + } } // Grabs last checksum for the named block file. @@ -1907,13 +1928,10 @@ func (fs *fileStore) expireMsgsOnRecover() { } // Make sure we do subject cleanup as well. mb.ensurePerSubjectInfoLoaded() - for subj := range mb.fss { - fs.removePerSubject(subj) - } - // Make sure we do subject cleanup as well. - mb.ensurePerSubjectInfoLoaded() - for subj := range mb.fss { - fs.removePerSubject(subj) + for subj, ss := range mb.fss { + for i := uint64(0); i < ss.Msgs; i++ { + fs.removePerSubject(subj) + } } mb.dirtyCloseWithRemove(true) deleted++ @@ -2044,7 +2062,7 @@ func (fs *fileStore) expireMsgsOnRecover() { lmb.writeTombstone(last.seq, last.ts) } // Clear any global subject state. - fs.psim = make(map[string]*psi) + fs.psim, fs.tsl = make(map[string]*psi), 0 } // If we purged anything, make sure we kick flush state loop. @@ -2408,7 +2426,7 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState { fs.mu.RLock() defer fs.mu.RUnlock() - if fs.state.Msgs == 0 { + if fs.state.Msgs == 0 || fs.noTrackSubjects() { return nil } @@ -2435,8 +2453,12 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState { } mb.mu.Lock() - // Make sure we have fss loaded. - mb.ensurePerSubjectInfoLoaded() + var shouldExpire bool + if mb.fss == nil { + // Make sure we have fss loaded. + mb.loadMsgsWithLock() + shouldExpire = true + } for subj, ss := range mb.fss { if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) { if ss.firstNeedsUpdate { @@ -2452,6 +2474,10 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState { } } } + if shouldExpire { + // Expire this cache before moving on. + mb.tryForceExpireCacheLocked() + } mb.mu.Unlock() if mb == stop { @@ -2943,6 +2969,7 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts in } } else { fs.psim[subj] = &psi{total: 1, fblk: index, lblk: index} + fs.tsl += len(subj) } } @@ -3190,14 +3217,16 @@ func (fs *fileStore) enforceBytesLimit() { // We will make sure to go through all msg blocks etc. but in practice this // will most likely only be the last one, so can take a more conservative approach. // Lock should be held. -func (fs *fileStore) enforceMsgPerSubjectLimit() { +func (fs *fileStore) enforceMsgPerSubjectLimit(fireCallback bool) { maxMsgsPer := uint64(fs.cfg.MaxMsgsPer) - // We want to suppress callbacks from remove during this process + // We may want to suppress callbacks from remove during this process // since these should have already been deleted and accounted for. - cb := fs.scb - fs.scb = nil - defer func() { fs.scb = cb }() + if !fireCallback { + cb := fs.scb + fs.scb = nil + defer func() { fs.scb = cb }() + } var numMsgs uint64 @@ -3215,7 +3244,7 @@ func (fs *fileStore) enforceMsgPerSubjectLimit() { if numMsgs != fs.state.Msgs { fs.warn("Detected skew in subject-based total (%d) vs raw total (%d), rebuilding", numMsgs, fs.state.Msgs) // Clear any global subject state. - fs.psim = make(map[string]*psi) + fs.psim, fs.tsl = make(map[string]*psi), 0 for _, mb := range fs.blks { ld, _, err := mb.rebuildState() if err != nil && ld != nil { @@ -3251,6 +3280,9 @@ func (fs *fileStore) enforceMsgPerSubjectLimit() { } // Grab the ss entry for this subject in case sparse. mb.mu.Lock() + if mb.cacheNotLoaded() { + mb.loadMsgsWithLock() + } mb.ensurePerSubjectInfoLoaded() ss := mb.fss[subj] if ss != nil && ss.firstNeedsUpdate { @@ -3321,6 +3353,7 @@ func (fs *fileStore) removePerSubject(subj string) { info.fblk = info.lblk } else if info.total == 0 { delete(fs.psim, subj) + fs.tsl -= len(subj) } } } @@ -3491,14 +3524,13 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) ( } else if !isEmpty { // Out of order delete. mb.dmap.Insert(seq) - // Check if <25% utilization and minimum size met. - if mb.rbytes > compactMinimum && !isLastBlock { - // Remove the interior delete records - rbytes := mb.rbytes - uint64(mb.dmap.Size()*emptyRecordLen) - if rbytes>>2 > mb.bytes { - mb.compact() - fs.kickFlushStateLoop() - } + // Make simple check here similar to Compact(). If we can save 50% and over a certain threshold do inline. + // All other more thorough cleanup will happen in syncBlocks logic. + // Note that we do not have to store empty records for the deleted, so don't use to calculate. + // TODO(dlc) - This should not be inline, should kick the sync routine. + if mb.rbytes > compactMinimum && mb.bytes*2 < mb.rbytes && !isLastBlock { + mb.compact() + fs.kickFlushStateLoop() } } @@ -3572,7 +3604,9 @@ func (mb *msgBlock) compact() { } buf := mb.cache.buf - nbuf := make([]byte, 0, len(buf)) + nbuf := getMsgBlockBuf(len(buf)) + // Recycle our nbuf when we are done. + defer recycleMsgBlockBuf(nbuf) var le = binary.LittleEndian var firstSet bool @@ -3622,9 +3656,16 @@ func (mb *msgBlock) compact() { } // Handle compression - var err error - if nbuf, err = mb.cmp.Compress(nbuf); err != nil { - return + if mb.cmp != NoCompression { + cbuf, err := mb.cmp.Compress(nbuf) + if err != nil { + return + } + meta := &CompressionInfo{ + Algorithm: mb.cmp, + OriginalSize: uint64(len(nbuf)), + } + nbuf = append(meta.MarshalMetadata(), cbuf...) } // Check for encryption. @@ -4258,7 +4299,7 @@ func (fs *fileStore) checkMsgs() *LostStreamData { fs.checkAndFlushAllBlocks() // Clear any global subject state. - fs.psim = make(map[string]*psi) + fs.psim, fs.tsl = make(map[string]*psi), 0 for _, mb := range fs.blks { // Make sure encryption loaded if needed for the block. @@ -4701,6 +4742,24 @@ func (mb *msgBlock) decompressIfNeeded(buf []byte) ([]byte, error) { } } +// Lock should be held. +func (mb *msgBlock) ensureRawBytesLoaded() error { + if mb.rbytes > 0 { + return nil + } + f, err := os.Open(mb.mfn) + if err != nil { + return err + } + defer f.Close() + if fi, err := f.Stat(); fi != nil && err == nil { + mb.rbytes = uint64(fi.Size()) + } else { + return err + } + return nil +} + // Sync msg and index files as needed. This is called from a timer. func (fs *fileStore) syncBlocks() { fs.mu.RLock() @@ -4709,8 +4768,10 @@ func (fs *fileStore) syncBlocks() { return } blks := append([]*msgBlock(nil), fs.blks...) + lmb := fs.lmb fs.mu.RUnlock() + var markDirty bool for _, mb := range blks { // Do actual sync. Hold lock for consistency. mb.mu.Lock() @@ -4722,24 +4783,33 @@ func (fs *fileStore) syncBlocks() { if mb.mfd != nil && mb.sinceLastWriteActivity() > closeFDsIdle { mb.dirtyCloseWithRemove(false) } + // Check if we should compact here as well. + // Do not compact last mb. + if mb != lmb && mb.ensureRawBytesLoaded() == nil && mb.rbytes > mb.bytes { + mb.compact() + markDirty = true + } + // Check if we need to sync. We will not hold lock during actual sync. - var fn string - if mb.needSync { + needSync, fn := mb.needSync, mb.mfn + if needSync { // Flush anything that may be pending. - if mb.pendingWriteSizeLocked() > 0 { - mb.flushPendingMsgsLocked() - } - fn = mb.mfn - mb.needSync = false + mb.flushPendingMsgsLocked() } mb.mu.Unlock() // Check if we need to sync. // This is done not holding any locks. - if fn != _EMPTY_ { + if needSync { if fd, _ := os.OpenFile(fn, os.O_RDWR, defaultFilePerms); fd != nil { - fd.Sync() + canClear := fd.Sync() == nil fd.Close() + // Only clear sync flag on success. + if canClear { + mb.mu.Lock() + mb.needSync = false + mb.mu.Unlock() + } } } } @@ -4748,8 +4818,12 @@ func (fs *fileStore) syncBlocks() { fs.syncTmr = time.AfterFunc(fs.fcfg.SyncInterval, fs.syncBlocks) fn := filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile) syncAlways := fs.fcfg.SyncAlways + if markDirty { + fs.dirty++ + } fs.mu.Unlock() + // Sync state file if we are not running with sync always. if !syncAlways { if fd, _ := os.OpenFile(fn, os.O_RDWR, defaultFilePerms); fd != nil { fd.Sync() @@ -5942,7 +6016,7 @@ func compareFn(subject string) func(string, string) bool { // Will return the number of purged messages. func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) { if subject == _EMPTY_ || subject == fwcs { - if keep == 0 && (sequence == 0 || sequence == 1) { + if keep == 0 && sequence == 0 { return fs.Purge() } if sequence > 1 { @@ -6098,7 +6172,7 @@ func (fs *fileStore) purge(fseq uint64) (uint64, error) { fs.lmb = nil fs.bim = make(map[uint32]*msgBlock) // Clear any per subject tracking. - fs.psim = make(map[string]*psi) + fs.psim, fs.tsl = make(map[string]*psi), 0 // Mark dirty fs.dirty++ @@ -6185,8 +6259,10 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) { bytes += mb.bytes // Make sure we do subject cleanup as well. mb.ensurePerSubjectInfoLoaded() - for subj := range mb.fss { - fs.removePerSubject(subj) + for subj, ss := range mb.fss { + for i := uint64(0); i < ss.Msgs; i++ { + fs.removePerSubject(subj) + } } // Now close. mb.dirtyCloseWithRemove(true) @@ -6369,7 +6445,7 @@ func (fs *fileStore) reset() error { fs.blks, fs.lmb = nil, nil // Reset subject mappings. - fs.psim = make(map[string]*psi) + fs.psim, fs.tsl = make(map[string]*psi), 0 fs.bim = make(map[uint32]*msgBlock) // If we purged anything, make sure we kick flush state loop. @@ -6629,7 +6705,12 @@ func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si var le = binary.LittleEndian for slot := startSlot; slot < len(mb.cache.idx); slot++ { - li := int(mb.cache.idx[slot]&^hbit) - mb.cache.off + bi := mb.cache.idx[slot] &^ hbit + if bi == dbit { + // delete marker so skip. + continue + } + li := int(bi) - mb.cache.off if li >= len(mb.cache.buf) { ss.First = ss.Last return @@ -6639,10 +6720,7 @@ func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si slen := int(le.Uint16(hdr[20:])) if subj == string(buf[msgHdrSize:msgHdrSize+slen]) { seq := le.Uint64(hdr[4:]) - if seq < mb.first.seq || seq&ebit != 0 { - continue - } - if mb.dmap.Exists(seq) { + if seq < mb.first.seq || seq&ebit != 0 || mb.dmap.Exists(seq) { continue } ss.First = seq @@ -6654,7 +6732,7 @@ func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si // Lock should be held. func (fs *fileStore) resetGlobalPerSubjectInfo() { // Clear any global subject state. - fs.psim = make(map[string]*psi) + fs.psim, fs.tsl = make(map[string]*psi), 0 for _, mb := range fs.blks { fs.populateGlobalPerSubjectInfo(mb) } @@ -6678,7 +6756,7 @@ func (mb *msgBlock) generatePerSubjectInfo() error { if err := mb.loadMsgsWithLock(); err != nil { return err } - // indexCaceheBuf can produce fss now, so if non-nil we are good. + // indexCacheBuf can produce fss now, so if non-nil we are good. if mb.fss != nil { return nil } @@ -6752,6 +6830,7 @@ func (fs *fileStore) populateGlobalPerSubjectInfo(mb *msgBlock) { } } else { fs.psim[subj] = &psi{total: ss.Msgs, fblk: mb.index, lblk: mb.index} + fs.tsl += len(subj) } } } @@ -6902,10 +6981,19 @@ func (fs *fileStore) writeFullState() error { return nil } - var _buf [32 * 1024]byte - _buf[0], _buf[1] = fullStateMagic, fullStateVersion - buf := _buf[:hdrLen] + // For calculating size. + numSubjects := len(fs.psim) + // Calculate and estimate of the uper bound on the size to avoid multiple allocations. + sz := 2 + // Magic and Version + (binary.MaxVarintLen64 * 6) + // FS data + binary.MaxVarintLen64 + fs.tsl + // NumSubjects + total subject length + numSubjects*(binary.MaxVarintLen64*4) + // psi record + len(fs.blks)*((binary.MaxVarintLen64*6)+512) + // msg blocks, 512 is est for dmap + binary.MaxVarintLen64 + 8 // last index + checksum + + buf := make([]byte, hdrLen, sz) + buf[0], buf[1] = fullStateMagic, fullStateVersion buf = binary.AppendUvarint(buf, fs.state.Msgs) buf = binary.AppendUvarint(buf, fs.state.Bytes) buf = binary.AppendUvarint(buf, fs.state.FirstSeq) @@ -6914,9 +7002,7 @@ func (fs *fileStore) writeFullState() error { buf = binary.AppendVarint(buf, timestampNormalized(fs.state.LastTime)) // Do per subject information map if applicable. - numSubjects := len(fs.psim) buf = binary.AppendUvarint(buf, uint64(numSubjects)) - if numSubjects > 0 { for subj, psi := range fs.psim { buf = binary.AppendUvarint(buf, uint64(len(subj))) @@ -6938,6 +7024,7 @@ func (fs *fileStore) writeFullState() error { // Use basetime to save some space. baseTime := timestampNormalized(fs.state.FirstTime) + var scratch [8 * 1024]byte for _, mb := range fs.blks { mb.mu.RLock() @@ -6951,7 +7038,6 @@ func (fs *fileStore) writeFullState() error { numDeleted := mb.dmap.Size() buf = binary.AppendUvarint(buf, uint64(numDeleted)) if numDeleted > 0 { - var scratch [8 * 1024]byte dmap, _ := mb.dmap.Encode(scratch[:0]) buf = append(buf, dmap...) } @@ -6991,6 +7077,10 @@ func (fs *fileStore) writeFullState() error { // Release lock. fs.mu.Unlock() + if cap(buf) > sz { + fs.warn("WriteFullState reallocated from %d to %d", sz, cap(buf)) + } + // Write to a tmp file and rename. const tmpPre = streamStreamStateFile + tsep f, err := os.CreateTemp(filepath.Join(fs.fcfg.StoreDir, msgDir), tmpPre) @@ -7270,6 +7360,9 @@ func (fs *fileStore) Snapshot(deadline time.Duration, checkMsgs, includeConsumer } } + // Write out full state as well before proceeding. + fs.writeFullState() + pr, pw := net.Pipe() // Set a write deadline here to protect ourselves. diff --git a/vendor/github.com/nats-io/nats-server/v2/server/jetstream.go b/vendor/github.com/nats-io/nats-server/v2/server/jetstream.go index 756e75a56..e029f4cdf 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/jetstream.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/jetstream.go @@ -1038,6 +1038,10 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro } s.mu.RLock() + if s.sys == nil { + s.mu.RUnlock() + return ErrServerNotRunning + } sendq := s.sys.sendq s.mu.RUnlock() diff --git a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_api.go b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_api.go index 7a3b9203c..c4163c853 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_api.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_api.go @@ -4120,7 +4120,9 @@ func (s *Server) jsConsumerListRequest(sub *subscription, c *client, _ *Account, } for _, o := range obs[offset:] { - resp.Consumers = append(resp.Consumers, o.info()) + if cinfo := o.info(); cinfo != nil { + resp.Consumers = append(resp.Consumers, cinfo) + } if len(resp.Consumers) >= JSApiListLimit { break } diff --git a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_cluster.go b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_cluster.go index 67278f20c..b5436facb 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_cluster.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_cluster.go @@ -2016,7 +2016,7 @@ func (js *jetStream) createRaftGroup(accName string, rg *raftGroup, storage Stor if storage == FileStorage { fs, err := newFileStoreWithCreated( FileStoreConfig{StoreDir: storeDir, BlockSize: defaultMediumBlockSize, AsyncFlush: false, SyncInterval: 5 * time.Minute, srv: s}, - StreamConfig{Name: rg.Name, Storage: FileStorage}, + StreamConfig{Name: rg.Name, Storage: FileStorage, Metadata: labels}, time.Now().UTC(), s.jsKeyGen(s.getOpts().JetStreamKey, rg.Name), s.jsKeyGen(s.getOpts().JetStreamOldKey, rg.Name), @@ -6472,7 +6472,7 @@ func (s *Server) jsClusteredStreamListRequest(acc *Account, ci *ClientInfo, filt s.sys.replies[inbox] = func(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) { var si StreamInfo if err := json.Unmarshal(msg, &si); err != nil { - s.Warnf("Error unmarshaling clustered stream info response:%v", err) + s.Warnf("Error unmarshalling clustered stream info response:%v", err) return } select { @@ -8121,7 +8121,7 @@ func (mset *stream) handleClusterSyncRequest(sub *subscription, c *client, _ *Ac func (js *jetStream) offlineClusterInfo(rg *raftGroup) *ClusterInfo { s := js.srv - ci := &ClusterInfo{Name: s.ClusterName()} + ci := &ClusterInfo{Name: s.ClusterName(), RaftGroup: rg.Name} for _, peer := range rg.Peers { if sir, ok := s.nodeToInfo.Load(peer); ok && sir != nil { si := sir.(nodeInfo) @@ -8150,8 +8150,9 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo { n := rg.node ci := &ClusterInfo{ - Name: s.cachedClusterName(), - Leader: s.serverNameForNode(n.GroupLeader()), + Name: s.cachedClusterName(), + Leader: s.serverNameForNode(n.GroupLeader()), + RaftGroup: rg.Name, } now := time.Now() diff --git a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_events.go b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_events.go index 2e1c7313a..ab7c61915 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_events.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_events.go @@ -134,6 +134,7 @@ type JSConsumerDeliveryTerminatedAdvisory struct { ConsumerSeq uint64 `json:"consumer_seq"` StreamSeq uint64 `json:"stream_seq"` Deliveries uint64 `json:"deliveries"` + Reason string `json:"reason,omitempty"` Domain string `json:"domain,omitempty"` } diff --git a/vendor/github.com/nats-io/nats-server/v2/server/memstore.go b/vendor/github.com/nats-io/nats-server/v2/server/memstore.go index 3ffaacafe..0d037be67 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/memstore.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/memstore.go @@ -427,9 +427,17 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje // We will adjust from the totals above by scanning what we need to exclude. ss.First = first var adjust uint64 + var tss *SimpleState + for seq := ms.state.FirstSeq; seq < first; seq++ { if sm, ok := ms.msgs[seq]; ok && !seen[sm.subj] && isMatch(sm.subj) { - adjust++ + if lastPerSubject { + tss = ms.fss[sm.subj] + } + // If we are last per subject, make sure to only adjust if all messages are before our first. + if tss == nil || tss.Last < first { + adjust++ + } if seen != nil { seen[sm.subj] = true } @@ -515,8 +523,7 @@ func (ms *memStore) SubjectsTotals(filterSubject string) map[string]uint64 { // NumPending will return the number of pending messages matching the filter subject starting at sequence. func (ms *memStore) NumPending(sseq uint64, filter string, lastPerSubject bool) (total, validThrough uint64) { - // This needs to be a write lock, as filteredStateLocked can - // mutate the per-subject state. + // This needs to be a write lock, as filteredStateLocked can mutate the per-subject state. ms.mu.Lock() defer ms.mu.Unlock() @@ -629,7 +636,7 @@ func (ms *memStore) expireMsgs() { // Will return the number of purged messages. func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) { if subject == _EMPTY_ || subject == fwcs { - if keep == 0 && (sequence == 0 || sequence == 1) { + if keep == 0 && sequence == 0 { return ms.Purge() } if sequence > 1 { diff --git a/vendor/github.com/nats-io/nats-server/v2/server/monitor.go b/vendor/github.com/nats-io/nats-server/v2/server/monitor.go index 66f5e81a3..073c468e0 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/monitor.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/monitor.go @@ -130,6 +130,7 @@ type ConnInfo struct { TLSVersion string `json:"tls_version,omitempty"` TLSCipher string `json:"tls_cipher_suite,omitempty"` TLSPeerCerts []*TLSPeerCert `json:"tls_peer_certs,omitempty"` + TLSFirst bool `json:"tls_first,omitempty"` AuthorizedUser string `json:"authorized_user,omitempty"` Account string `json:"account,omitempty"` Subs []string `json:"subscriptions_list,omitempty"` @@ -568,6 +569,7 @@ func (ci *ConnInfo) fill(client *client, nc net.Conn, now time.Time, auth bool) if auth && len(cs.PeerCertificates) > 0 { ci.TLSPeerCerts = makePeerCerts(cs.PeerCertificates) } + ci.TLSFirst = client.flags.isSet(didTLSFirst) } } diff --git a/vendor/github.com/nats-io/nats-server/v2/server/mqtt.go b/vendor/github.com/nats-io/nats-server/v2/server/mqtt.go index c0d17f26f..730272200 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/mqtt.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/mqtt.go @@ -85,6 +85,7 @@ const ( mqttConnAckRCServerUnavailable = byte(0x3) mqttConnAckRCBadUserOrPassword = byte(0x4) mqttConnAckRCNotAuthorized = byte(0x5) + mqttConnAckRCQoS2WillRejected = byte(0x10) // Maximum payload size of a control packet mqttMaxPayloadSize = 0xFFFFFFF @@ -345,6 +346,14 @@ type mqtt struct { asm *mqttAccountSessionManager // quick reference to account session manager, immutable after processConnect() sess *mqttSession // quick reference to session, immutable after processConnect() cid string // client ID + + // rejectQoS2Pub tells the MQTT client to not accept QoS2 PUBLISH, instead + // error and terminate the connection. + rejectQoS2Pub bool + + // downgradeQOS2Sub tells the MQTT client to downgrade QoS2 SUBSCRIBE + // requests to QoS1. + downgradeQoS2Sub bool } type mqttPending struct { @@ -475,7 +484,11 @@ func (s *Server) createMQTTClient(conn net.Conn, ws *websocket) *client { } now := time.Now() - c := &client{srv: s, nc: conn, mpay: maxPay, msubs: maxSubs, start: now, last: now, mqtt: &mqtt{}, ws: ws} + mqtt := &mqtt{ + rejectQoS2Pub: opts.MQTT.rejectQoS2Pub, + downgradeQoS2Sub: opts.MQTT.downgradeQoS2Sub, + } + c := &client{srv: s, nc: conn, mpay: maxPay, msubs: maxSubs, start: now, last: now, mqtt: mqtt, ws: ws} c.headers = true c.mqtt.pp = &mqttPublish{} // MQTT clients don't send NATS CONNECT protocols. So make it an "echo" @@ -598,9 +611,9 @@ func validateMQTTOptions(o *Options) error { if mo.Port == 0 { return nil } - // We have to force the server name to be explicitly set. There are conditions - // where we need a unique, repeatable name. - if o.ServerName == _EMPTY_ { + // We have to force the server name to be explicitly set and be unique when + // in cluster mode. + if o.ServerName == _EMPTY_ && (o.Cluster.Port != 0 || o.Gateway.Port != 0) { return errMQTTServerNameMustBeSet } // If there is a NoAuthUser, we need to have Users defined and @@ -2029,9 +2042,6 @@ func (as *mqttAccountSessionManager) sendJSAPIrequests(s *Server, c *client, acc // Add/Replace this message from the retained messages map. // If a message for this topic already existed, the existing record is updated // with the provided information. -// This function will return the stream sequence of the record before its update, -// or 0 if the record was added instead of updated. -// // Lock not held on entry. func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetainedMsgRef) { as.mu.Lock() @@ -2040,7 +2050,7 @@ func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetai as.retmsgs = make(map[string]*mqttRetainedMsgRef) as.sl = NewSublistWithCache() } else { - // Check if we already had one. If so, update the existing one. + // Check if we already had one retained message. If so, update the existing one. if erm, exists := as.retmsgs[key]; exists { // If the new sequence is below the floor or the existing one, // then ignore the new one. @@ -2057,6 +2067,7 @@ func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetai erm.sub = &subscription{subject: []byte(key)} as.sl.Insert(erm.sub) } + return } } rm.sub = &subscription{subject: []byte(key)} @@ -2228,6 +2239,10 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client, if f.qos > 2 { f.qos = 2 } + if c.mqtt.downgradeQoS2Sub && f.qos == 2 { + c.Warnf("Downgrading subscription QoS2 to QoS1 for %q, as configured", f.filter) + f.qos = 1 + } subject := f.filter sid := subject @@ -2331,6 +2346,10 @@ func (as *mqttAccountSessionManager) serializeRetainedMsgsForSub(sess *mqttSessi if qos > sub.mqtt.qos { qos = sub.mqtt.qos } + if c.mqtt.rejectQoS2Pub && qos == 2 { + c.Warnf("Rejecting retained message with QoS2 for subscription %q, as configured", sub.subject) + continue + } if qos > 0 { pi = sess.trackPublishRetained() @@ -2891,10 +2910,11 @@ func (sess *mqttSession) trackAsPubRel(pi uint16, jsAckSubject string) { sseq, _, _ := ackReplyInfo(jsAckSubject) - var sseqToPi map[uint64]uint16 if sess.cpending == nil { sess.cpending = make(map[string]map[uint64]uint16) - } else if sseqToPi = sess.cpending[jsDur]; sseqToPi == nil { + } + sseqToPi := sess.cpending[jsDur] + if sseqToPi == nil { sseqToPi = make(map[uint64]uint16) sess.cpending[jsDur] = sseqToPi } @@ -3002,6 +3022,10 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int, hasMappings bool) (byte hasWill = true } + if c.mqtt.rejectQoS2Pub && hasWill && wqos == 2 { + return mqttConnAckRCQoS2WillRejected, nil, fmt.Errorf("server does not accept QoS2 for Will messages") + } + // Spec [MQTT-3.1.2-19] hasUser := cp.flags&mqttConnFlagUsernameFlag != 0 // Spec [MQTT-3.1.2-21] @@ -3386,6 +3410,11 @@ func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish, hasMapping if qos > 2 { return fmt.Errorf("QoS=%v is invalid in MQTT", qos) } + + if c.mqtt.rejectQoS2Pub && qos == 2 { + return fmt.Errorf("QoS=2 is disabled for PUBLISH messages") + } + // Keep track of where we are when starting to read the variable header start := r.pos diff --git a/vendor/github.com/nats-io/nats-server/v2/server/opts.go b/vendor/github.com/nats-io/nats-server/v2/server/opts.go index 039f982c0..ff46afb0d 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/opts.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/opts.go @@ -327,11 +327,23 @@ type Options struct { TLSConfig *tls.Config `json:"-"` TLSPinnedCerts PinnedCertSet `json:"-"` TLSRateLimit int64 `json:"-"` - AllowNonTLS bool `json:"-"` - WriteDeadline time.Duration `json:"-"` - MaxClosedClients int `json:"-"` - LameDuckDuration time.Duration `json:"-"` - LameDuckGracePeriod time.Duration `json:"-"` + // When set to true, the server will perform the TLS handshake before + // sending the INFO protocol. For clients that are not configured + // with a similar option, their connection will fail with some sort + // of timeout or EOF error since they are expecting to receive an + // INFO protocol first. + TLSHandshakeFirst bool `json:"-"` + // If TLSHandshakeFirst is true and this value is strictly positive, + // the server will wait for that amount of time for the TLS handshake + // to start before falling back to previous behavior of sending the + // INFO protocol first. It allows for a mix of newer clients that can + // require a TLS handshake first, and older clients that can't. + TLSHandshakeFirstFallback time.Duration `json:"-"` + AllowNonTLS bool `json:"-"` + WriteDeadline time.Duration `json:"-"` + MaxClosedClients int `json:"-"` + LameDuckDuration time.Duration `json:"-"` + LameDuckGracePeriod time.Duration `json:"-"` // MaxTracedMsgLen is the maximum printable length for traced messages. MaxTracedMsgLen int `json:"-"` @@ -550,6 +562,14 @@ type MQTTOpts struct { // Snapshot of configured TLS options. tlsConfigOpts *TLSConfigOpts + + // rejectQoS2Pub tells the MQTT client to not accept QoS2 PUBLISH, instead + // error and terminate the connection. + rejectQoS2Pub bool + + // downgradeQOS2Sub tells the MQTT client to downgrade QoS2 SUBSCRIBE + // requests to QoS1. + downgradeQoS2Sub bool } type netResolver interface { @@ -638,7 +658,8 @@ type TLSConfigOpts struct { Insecure bool Map bool TLSCheckKnownURLs bool - HandshakeFirst bool // Indicate that the TLS handshake should occur first, before sending the INFO protocol + HandshakeFirst bool // Indicate that the TLS handshake should occur first, before sending the INFO protocol. + FallbackDelay time.Duration // Where supported, indicates how long to wait for the handshake before falling back to sending the INFO protocol first. Timeout float64 RateLimit int64 Ciphers []uint16 @@ -1072,6 +1093,8 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error o.TLSMap = tc.Map o.TLSPinnedCerts = tc.PinnedCerts o.TLSRateLimit = tc.RateLimit + o.TLSHandshakeFirst = tc.HandshakeFirst + o.TLSHandshakeFirstFallback = tc.FallbackDelay // Need to keep track of path of the original TLS config // and certs path for OCSP Stapling monitoring. @@ -4312,7 +4335,30 @@ func parseTLS(v interface{}, isClientCtx bool) (t *TLSConfigOpts, retErr error) } tc.CertMatch = certMatch case "handshake_first", "first", "immediate": - tc.HandshakeFirst = mv.(bool) + switch mv := mv.(type) { + case bool: + tc.HandshakeFirst = mv + case string: + switch strings.ToLower(mv) { + case "true", "on": + tc.HandshakeFirst = true + case "false", "off": + tc.HandshakeFirst = false + case "auto", "auto_fallback": + tc.HandshakeFirst = true + tc.FallbackDelay = DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY + default: + // Check to see if this is a duration. + if dur, err := time.ParseDuration(mv); err == nil { + tc.HandshakeFirst = true + tc.FallbackDelay = dur + break + } + return nil, &configErr{tk, fmt.Sprintf("field %q's value %q is invalid", mk, mv)} + } + default: + return nil, &configErr{tk, fmt.Sprintf("field %q should be a boolean or a string, got %T", mk, mv)} + } case "ocsp_peer": switch vv := mv.(type) { case bool: @@ -4593,6 +4639,11 @@ func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) er case "consumer_inactive_threshold", "consumer_auto_cleanup": o.MQTT.ConsumerInactiveThreshold = parseDuration("consumer_inactive_threshold", tk, mv, errors, warnings) + case "reject_qos2_publish": + o.MQTT.rejectQoS2Pub = mv.(bool) + case "downgrade_qos2_subscribe": + o.MQTT.downgradeQoS2Sub = mv.(bool) + default: if !tk.IsUsedVariable() { err := &unknownConfigFieldErr{ diff --git a/vendor/github.com/nats-io/nats-server/v2/server/raft.go b/vendor/github.com/nats-io/nats-server/v2/server/raft.go index 7baa949df..b0f30786b 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/raft.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/raft.go @@ -133,8 +133,7 @@ type raft struct { wtype StorageType track bool werr error - state RaftState - isLeader atomic.Bool + state atomic.Int32 // RaftState hh hash.Hash64 snapfile string csz int @@ -382,7 +381,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe wal: cfg.Log, wtype: cfg.Log.Type(), track: cfg.Track, - state: Follower, csz: ps.clusterSize, qn: ps.clusterSize/2 + 1, hash: hash, @@ -653,12 +651,11 @@ func (s *Server) transferRaftLeaders() bool { // Propose will propose a new entry to the group. // This should only be called on the leader. func (n *raft) Propose(data []byte) error { - n.RLock() - if state := n.state; state != Leader { - n.RUnlock() + if state := n.State(); state != Leader { n.debug("Proposal ignored, not leader (state: %v)", state) return errNotLeader } + n.RLock() // Error if we had a previous write error. if werr := n.werr; werr != nil { n.RUnlock() @@ -674,12 +671,11 @@ func (n *raft) Propose(data []byte) error { // ProposeDirect will propose entries directly. // This should only be called on the leader. func (n *raft) ProposeDirect(entries []*Entry) error { - n.RLock() - if n.state != Leader { - n.RUnlock() - n.debug("Direct proposal ignored, not leader (state: %v)", n.state) + if state := n.State(); state != Leader { + n.debug("Direct proposal ignored, not leader (state: %v)", state) return errNotLeader } + n.RLock() // Error if we had a previous write error. if werr := n.werr; werr != nil { n.RUnlock() @@ -706,11 +702,10 @@ func (n *raft) ForwardProposal(entry []byte) error { // ProposeAddPeer is called to add a peer to the group. func (n *raft) ProposeAddPeer(peer string) error { - n.RLock() - if n.state != Leader { - n.RUnlock() + if n.State() != Leader { return errNotLeader } + n.RLock() // Error if we had a previous write error. if werr := n.werr; werr != nil { n.RUnlock() @@ -742,7 +737,7 @@ func (n *raft) doRemovePeerAsLeader(peer string) { func (n *raft) ProposeRemovePeer(peer string) error { n.RLock() prop, subj := n.prop, n.rpsubj - isLeader := n.state == Leader + isLeader := n.State() == Leader werr := n.werr n.RUnlock() @@ -794,11 +789,10 @@ func (n *raft) AdjustBootClusterSize(csz int) error { // AdjustClusterSize will change the cluster set size. // Must be the leader. func (n *raft) AdjustClusterSize(csz int) error { - n.Lock() - if n.state != Leader { - n.Unlock() + if n.State() != Leader { return errNotLeader } + n.Lock() // Same floor as bootstrap. if csz < 2 { csz = 2 @@ -816,14 +810,15 @@ func (n *raft) AdjustClusterSize(csz int) error { // PauseApply will allow us to pause processing of append entries onto our // external apply chan. func (n *raft) PauseApply() error { + if n.State() == Leader { + return errAlreadyLeader + } + n.Lock() defer n.Unlock() - if n.state == Leader { - return errAlreadyLeader - } // If we are currently a candidate make sure we step down. - if n.state == Candidate { + if n.State() == Candidate { n.stepdown.push(noLeader) } @@ -944,12 +939,12 @@ func (n *raft) SendSnapshot(data []byte) error { // all of the log entries up to and including index. This should not be called with // entries that have been applied to the FSM but have not been applied to the raft state. func (n *raft) InstallSnapshot(data []byte) error { - n.Lock() - if n.state == Closed { - n.Unlock() + if n.State() == Closed { return errNodeClosed } + n.Lock() + if werr := n.werr; werr != nil { n.Unlock() return werr @@ -1164,7 +1159,7 @@ func (n *raft) Leader() bool { if n == nil { return false } - return n.isLeader.Load() + return n.State() == Leader } func (n *raft) isCatchingUp() bool { @@ -1178,7 +1173,7 @@ func (n *raft) isCatchingUp() bool { // Lock should be held. func (n *raft) isCurrent(includeForwardProgress bool) bool { // Check if we are closed. - if n.state == Closed { + if n.State() == Closed { n.debug("Not current, node is closed") return false } @@ -1190,7 +1185,7 @@ func (n *raft) isCurrent(includeForwardProgress bool) bool { } // Make sure we are the leader or we know we have heard from the leader recently. - if n.state == Leader { + if n.State() == Leader { return true } @@ -1300,7 +1295,7 @@ func (n *raft) StepDown(preferred ...string) error { return errTooManyPrefs } - if n.state != Leader { + if n.State() != Leader { n.Unlock() return errNotLeader } @@ -1394,7 +1389,7 @@ func randCampaignTimeout() time.Duration { // Lock should be held. func (n *raft) campaign() error { n.debug("Starting campaign") - if n.state == Leader { + if n.State() == Leader { return errAlreadyLeader } n.resetElect(randCampaignTimeout()) @@ -1405,7 +1400,7 @@ func (n *raft) campaign() error { // Lock should be held. func (n *raft) xferCampaign() error { n.debug("Starting transfer campaign") - if n.state == Leader { + if n.State() == Leader { n.lxfer = false return errAlreadyLeader } @@ -1415,9 +1410,7 @@ func (n *raft) xferCampaign() error { // State returns the current state for this node. func (n *raft) State() RaftState { - n.RLock() - defer n.RUnlock() - return n.state + return RaftState(n.state.Load()) } // Progress returns the current index, commit and applied values. @@ -1478,7 +1471,7 @@ func (n *raft) UpdateKnownPeers(knownPeers []string) { // Process like peer state update. ps := &peerState{knownPeers, len(knownPeers), n.extSt} n.processPeerState(ps) - isLeader := n.state == Leader + isLeader := n.State() == Leader n.Unlock() // If we are the leader send this update out as well. @@ -1507,7 +1500,14 @@ func (n *raft) Delete() { func (n *raft) shutdown(shouldDelete bool) { n.Lock() - if n.state == Closed { + + // Returned swap value is the previous state. It looks counter-intuitive + // to do this atomic operation with the lock held, but we have to do so in + // order to make sure that switchState() is not already running. If it is + // then it can potentially update the n.state back to a non-closed state, + // allowing shutdown() to be called again. If that happens then the below + // close(n.quit) will panic from trying to close an already-closed channel. + if n.state.Swap(int32(Closed)) == int32(Closed) { n.Unlock() return } @@ -1525,7 +1525,6 @@ func (n *raft) shutdown(shouldDelete bool) { } c.closeConnection(InternalClient) } - n.state = Closed s, g, wal := n.s, n.group, n.wal // Delete our peer state and vote state and any snapshots. @@ -1788,7 +1787,7 @@ func (n *raft) processAppendEntries() { } func (n *raft) runAsFollower() { - for { + for n.State() == Follower { elect := n.electTimer() select { @@ -2156,11 +2155,11 @@ func (n *raft) handleForwardedProposal(sub *subscription, c *client, _ *Account, } func (n *raft) runAsLeader() { - n.RLock() - if n.state == Closed { - n.RUnlock() + if n.State() == Closed { return } + + n.RLock() psubj, rpsubj := n.psubj, n.rpsubj n.RUnlock() @@ -2193,7 +2192,7 @@ func (n *raft) runAsLeader() { lq := time.NewTicker(lostQuorumCheck) defer lq.Stop() - for { + for n.State() == Leader { select { case <-n.s.quitCh: n.shutdown(false) @@ -2541,7 +2540,7 @@ func (n *raft) loadEntry(index uint64) (*appendEntry, error) { // applyCommit will update our commit index and apply the entry to the apply chan. // lock should be held. func (n *raft) applyCommit(index uint64) error { - if n.state == Closed { + if n.State() == Closed { return errNodeClosed } if index <= n.commit { @@ -2551,7 +2550,7 @@ func (n *raft) applyCommit(index uint64) error { original := n.commit n.commit = index - if n.state == Leader { + if n.State() == Leader { delete(n.acks, index) } @@ -2568,7 +2567,7 @@ func (n *raft) applyCommit(index uint64) error { if ae, err = n.loadEntry(index); err != nil { if err != ErrStoreClosed && err != ErrStoreEOF { n.warn("Got an error loading %d index: %v - will reset", index, err) - if n.state == Leader { + if n.State() == Leader { n.stepdown.push(n.selectNextLeader()) } // Reset and cancel any catchup. @@ -2596,7 +2595,7 @@ func (n *raft) applyCommit(index uint64) error { case EntrySnapshot: committed = append(committed, e) case EntryPeerState: - if n.state != Leader { + if n.State() != Leader { if ps, err := decodePeerState(e.Data); err == nil { n.processPeerState(ps) } @@ -2646,7 +2645,7 @@ func (n *raft) applyCommit(index uint64) error { } // If this is us and we are the leader we should attempt to stepdown. - if peer == n.id && n.state == Leader { + if peer == n.id && n.State() == Leader { n.stepdown.push(n.selectNextLeader()) } @@ -2674,12 +2673,12 @@ func (n *raft) applyCommit(index uint64) error { // Used to track a success response and apply entries. func (n *raft) trackResponse(ar *appendEntryResponse) { - n.Lock() - if n.state == Closed { - n.Unlock() + if n.State() == Closed { return } + n.Lock() + // Update peer's last index. if ps := n.peers[ar.peer]; ps != nil && ar.index > ps.li { ps.li = ar.index @@ -2736,7 +2735,7 @@ func (n *raft) adjustClusterSizeAndQuorum() { n.lsut = time.Now() } else if ncsz < pcsz { n.debug("Decreasing our clustersize: %d -> %d", pcsz, ncsz) - if n.state == Leader { + if n.State() == Leader { go n.sendHeartbeat() } } @@ -2749,7 +2748,7 @@ func (n *raft) trackPeer(peer string) error { if n.removed != nil { _, isRemoved = n.removed[peer] } - if n.state == Leader { + if n.State() == Leader { if lp, ok := n.peers[peer]; !ok || !lp.kp { // Check if this peer had been removed previously. needPeerAdd = !isRemoved @@ -2780,7 +2779,7 @@ func (n *raft) runAsCandidate() { // We vote for ourselves. votes := 1 - for { + for n.State() == Candidate { elect := n.electTimer() select { case <-n.entry.ch: @@ -2962,7 +2961,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) { } // Just return if closed or we had previous write error. - if n.state == Closed || n.werr != nil { + if n.State() == Closed || n.werr != nil { n.Unlock() return } @@ -2972,7 +2971,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) { arbuf := scratch[:] // Are we receiving from another leader. - if n.state == Leader { + if n.State() == Leader { // If we are the same we should step down to break the tie. if ae.term >= n.term { n.term = ae.term @@ -2993,7 +2992,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) { } // If we received an append entry as a candidate we should convert to a follower. - if n.state == Candidate { + if n.State() == Candidate { n.debug("Received append entry in candidate state from %q, converting to follower", ae.leader) if n.term < ae.term { n.term = ae.term @@ -3059,13 +3058,13 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) { if isNew { n.writeTermVote() } - if n.state != Follower { - n.debug("Term higher than ours and we are not a follower: %v, stepping down to %q", n.state, ae.leader) + if n.State() != Follower { + n.debug("Term higher than ours and we are not a follower: %v, stepping down to %q", n.State(), ae.leader) n.stepdown.push(ae.leader) } } - if isNew && n.leader != ae.leader && n.state == Follower { + if isNew && n.leader != ae.leader && n.State() == Follower { n.debug("AppendEntry updating leader to %q", ae.leader) n.updateLeader(ae.leader) n.writeTermVote() @@ -3333,7 +3332,7 @@ func (n *raft) storeToWAL(ae *appendEntry) error { // Sanity checking for now. if index := ae.pindex + 1; index != seq { n.warn("Wrong index, ae is %+v, index stored was %d, n.pindex is %d, will reset", ae, seq, n.pindex) - if n.state == Leader { + if n.State() == Leader { n.stepdown.push(n.selectNextLeader()) } // Reset and cancel any catchup. @@ -3564,7 +3563,7 @@ func (n *raft) readTermVote() (term uint64, voted string, err error) { // Lock should be held. func (n *raft) setWriteErrLocked(err error) { // Check if we are closed already. - if n.state == Closed { + if n.State() == Closed { return } // Ignore if already set. @@ -3598,9 +3597,7 @@ func (n *raft) setWriteErrLocked(err error) { // Helper to check if we are closed when we do not hold a lock already. func (n *raft) isClosed() bool { - n.RLock() - defer n.RUnlock() - return n.state == Closed + return n.State() == Closed } // Capture our write error if any and hold. @@ -3746,9 +3743,9 @@ func (n *raft) processVoteRequest(vr *voteRequest) error { // If this is a higher term go ahead and stepdown. if vr.term > n.term { - if n.state != Follower { + if n.State() != Follower { n.debug("Stepping down from %s, detected higher term: %d vs %d", - strings.ToLower(n.state.String()), vr.term, n.term) + strings.ToLower(n.State().String()), vr.term, n.term) n.stepdown.push(noLeader) n.term = vr.term } @@ -3787,7 +3784,7 @@ func (n *raft) handleVoteRequest(sub *subscription, c *client, _ *Account, subje func (n *raft) requestVote() { n.Lock() - if n.state != Candidate { + if n.State() != Candidate { n.Unlock() return } @@ -3829,9 +3826,6 @@ func (n *raft) quorumNeeded() int { // Lock should be held. func (n *raft) updateLeadChange(isLeader bool) { - // Update our atomic about being the leader. - n.isLeader.Store(isLeader) - // We don't care about values that have not been consumed (transitory states), // so we dequeue any state that is pending and push the new one. for { @@ -3851,25 +3845,25 @@ func (n *raft) updateLeadChange(isLeader bool) { // Lock should be held. func (n *raft) switchState(state RaftState) { - if n.state == Closed { + if n.State() == Closed { return } // Reset the election timer. n.resetElectionTimeout() - if n.state == Leader && state != Leader { + if n.State() == Leader && state != Leader { n.updateLeadChange(false) // Drain the response queue. n.resp.drain() - } else if state == Leader && n.state != Leader { + } else if state == Leader && n.State() != Leader { if len(n.pae) > 0 { n.pae = make(map[uint64]*appendEntry) } n.updateLeadChange(true) } - n.state = state + n.state.Store(int32(state)) n.writeTermVote() } @@ -3879,11 +3873,13 @@ const ( ) func (n *raft) switchToFollower(leader string) { - n.Lock() - defer n.Unlock() - if n.state == Closed { + if n.State() == Closed { return } + + n.Lock() + defer n.Unlock() + n.debug("Switching to follower") n.lxfer = false @@ -3892,17 +3888,19 @@ func (n *raft) switchToFollower(leader string) { } func (n *raft) switchToCandidate() { - n.Lock() - defer n.Unlock() - if n.state == Closed { + if n.State() == Closed { return } + + n.Lock() + defer n.Unlock() + // If we are catching up or are in observer mode we can not switch. if n.observer || n.paused { return } - if n.state != Candidate { + if n.State() != Candidate { n.debug("Switching to candidate") } else { if n.lostQuorumLocked() && time.Since(n.llqrt) > 20*time.Second { @@ -3919,11 +3917,12 @@ func (n *raft) switchToCandidate() { } func (n *raft) switchToLeader() { - n.Lock() - if n.state == Closed { - n.Unlock() + if n.State() == Closed { return } + + n.Lock() + n.debug("Switching to leader") var state StreamState diff --git a/vendor/github.com/nats-io/nats-server/v2/server/reload.go b/vendor/github.com/nats-io/nats-server/v2/server/reload.go index 239881715..4e1c2f71b 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/reload.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/reload.go @@ -266,6 +266,28 @@ func (t *tlsPinnedCertOption) Apply(server *Server) { server.Noticef("Reloaded: %d pinned_certs", len(t.newValue)) } +// tlsHandshakeFirst implements the option interface for the tls `handshake first` setting. +type tlsHandshakeFirst struct { + noopOption + newValue bool +} + +// Apply is a no-op because the timeout will be reloaded after options are applied. +func (t *tlsHandshakeFirst) Apply(server *Server) { + server.Noticef("Reloaded: Client TLS handshake first: %v", t.newValue) +} + +// tlsHandshakeFirstFallback implements the option interface for the tls `handshake first fallback delay` setting. +type tlsHandshakeFirstFallback struct { + noopOption + newValue time.Duration +} + +// Apply is a no-op because the timeout will be reloaded after options are applied. +func (t *tlsHandshakeFirstFallback) Apply(server *Server) { + server.Noticef("Reloaded: Client TLS handshake first fallback delay: %v", t.newValue) +} + // authOption is a base struct that provides default option behaviors. type authOption struct { noopOption @@ -1222,6 +1244,10 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) { diffOpts = append(diffOpts, &tlsTimeoutOption{newValue: newValue.(float64)}) case "tlspinnedcerts": diffOpts = append(diffOpts, &tlsPinnedCertOption{newValue: newValue.(PinnedCertSet)}) + case "tlshandshakefirst": + diffOpts = append(diffOpts, &tlsHandshakeFirst{newValue: newValue.(bool)}) + case "tlshandshakefirstfallback": + diffOpts = append(diffOpts, &tlsHandshakeFirstFallback{newValue: newValue.(time.Duration)}) case "username": diffOpts = append(diffOpts, &usernameOption{}) case "password": diff --git a/vendor/github.com/nats-io/nats-server/v2/server/route.go b/vendor/github.com/nats-io/nats-server/v2/server/route.go index 493f03e47..4cada7c8d 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/route.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/route.go @@ -1071,9 +1071,30 @@ func (s *Server) processImplicitRoute(info *Info, routeNoPool bool) { // in the server's opts.Routes, false otherwise. // Server lock is assumed to be held by caller. func (s *Server) hasThisRouteConfigured(info *Info) bool { - urlToCheckExplicit := strings.ToLower(net.JoinHostPort(info.Host, strconv.Itoa(info.Port))) - for _, ri := range s.getOpts().Routes { - if strings.ToLower(ri.Host) == urlToCheckExplicit { + routes := s.getOpts().Routes + if len(routes) == 0 { + return false + } + // This could possibly be a 0.0.0.0 host so we will also construct a second + // url with the host section of the `info.IP` (if present). + sPort := strconv.Itoa(info.Port) + urlOne := strings.ToLower(net.JoinHostPort(info.Host, sPort)) + var urlTwo string + if info.IP != _EMPTY_ { + if u, _ := url.Parse(info.IP); u != nil { + urlTwo = strings.ToLower(net.JoinHostPort(u.Hostname(), sPort)) + // Ignore if same than the first + if urlTwo == urlOne { + urlTwo = _EMPTY_ + } + } + } + for _, ri := range routes { + rHost := strings.ToLower(ri.Host) + if rHost == urlOne { + return true + } + if urlTwo != _EMPTY_ && rHost == urlTwo { return true } } @@ -1860,6 +1881,17 @@ const ( func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string) bool { id := info.ID + var acc *Account + if accName != _EMPTY_ { + var err error + acc, err = s.LookupAccount(accName) + if err != nil { + c.sendErrAndErr(fmt.Sprintf("Unable to lookup account %q: %v", accName, err)) + c.closeConnection(MissingAccount) + return false + } + } + s.mu.Lock() if !s.isRunning() || s.routesReject { s.mu.Unlock() @@ -1935,6 +1967,9 @@ func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string if c.last.IsZero() { c.last = time.Now() } + if acc != nil { + c.acc = acc + } c.mu.Unlock() // Store this route with key being the route id hash + account name @@ -1984,6 +2019,21 @@ func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string s.mu.Unlock() return false } + // Look if there is a solicited route in the pool. If there is one, + // they should all be, so stop at the first. + if url, rtype, hasSolicited := hasSolicitedRoute(conns); hasSolicited { + upgradeRouteToSolicited(c, url, rtype) + } + } else { + // If we solicit, upgrade to solicited all non-solicited routes that + // we may have registered. + c.mu.Lock() + url := c.route.url + rtype := c.route.routeType + c.mu.Unlock() + for _, r := range conns { + upgradeRouteToSolicited(r, url, rtype) + } } // For all cases (solicited and not) we need to count how many connections // we already have, and for solicited route, we will find a free spot in @@ -2109,6 +2159,41 @@ func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string return !exists } +func hasSolicitedRoute(conns []*client) (*url.URL, RouteType, bool) { + var url *url.URL + var rtype RouteType + for _, r := range conns { + if r == nil { + continue + } + r.mu.Lock() + if r.route.didSolicit { + url = r.route.url + rtype = r.route.routeType + } + r.mu.Unlock() + if url != nil { + return url, rtype, true + } + } + return nil, 0, false +} + +func upgradeRouteToSolicited(r *client, url *url.URL, rtype RouteType) { + if r == nil { + return + } + r.mu.Lock() + if !r.route.didSolicit { + r.route.didSolicit = true + r.route.url = url + } + if rtype == Explicit { + r.route.routeType = Explicit + } + r.mu.Unlock() +} + func handleDuplicateRoute(remote, c *client, setNoReconnect bool) { // We used to clear some fields when closing a duplicate connection // to prevent sending INFO protocols for the remotes to update diff --git a/vendor/github.com/nats-io/nats-server/v2/server/server.go b/vendor/github.com/nats-io/nats-server/v2/server/server.go index d1d0d109d..b11389030 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/server.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/server.go @@ -2573,6 +2573,9 @@ func (s *Server) AcceptLoop(clr chan struct{}) { // Alert of TLS enabled. if opts.TLSConfig != nil { s.Noticef("TLS required for client connections") + if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 { + s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect") + } } // If server was started with RANDOM_PORT (-1), opts.Port would be equal @@ -3041,10 +3044,37 @@ func (s *Server) createClientEx(conn net.Conn, inProcess bool) *client { c.Debugf("Client connection created") - // Send our information. - // Need to be sent in place since writeLoop cannot be started until - // TLS handshake is done (if applicable). - c.sendProtoNow(c.generateClientInfoJSON(info)) + // Save info.TLSRequired value since we may neeed to change it back and forth. + orgInfoTLSReq := info.TLSRequired + + var tlsFirstFallback time.Duration + // Check if we should do TLS first. + tlsFirst := opts.TLSConfig != nil && opts.TLSHandshakeFirst + if tlsFirst { + // Make sure info.TLSRequired is set to true (it could be false + // if AllowNonTLS is enabled). + info.TLSRequired = true + // Get the fallback delay value if applicable. + if f := opts.TLSHandshakeFirstFallback; f > 0 { + tlsFirstFallback = f + } else if inProcess { + // For in-process connection, we will always have a fallback + // delay. It allows support for non-TLS, TLS and "TLS First" + // in-process clients to successfully connect. + tlsFirstFallback = DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY + } + } + + // Decide if we are going to require TLS or not and generate INFO json. + tlsRequired := info.TLSRequired + infoBytes := c.generateClientInfoJSON(info) + + // Send our information, except if TLS and TLSHandshakeFirst is requested. + if !tlsFirst { + // Need to be sent in place since writeLoop cannot be started until + // TLS handshake is done (if applicable). + c.sendProtoNow(infoBytes) + } // Unlock to register c.mu.Unlock() @@ -3077,20 +3107,50 @@ func (s *Server) createClientEx(conn net.Conn, inProcess bool) *client { } s.clients[c.cid] = c - tlsRequired := info.TLSRequired s.mu.Unlock() // Re-Grab lock c.mu.Lock() - // Connection could have been closed while sending the INFO proto. isClosed := c.isClosed() - var pre []byte + // We need first to check for "TLS First" fallback delay. + if !isClosed && tlsFirstFallback > 0 { + // We wait and see if we are getting any data. Since we did not send + // the INFO protocol yet, only clients that use TLS first should be + // sending data (the TLS handshake). We don't really check the content: + // if it is a rogue agent and not an actual client performing the + // TLS handshake, the error will be detected when performing the + // handshake on our side. + pre = make([]byte, 4) + c.nc.SetReadDeadline(time.Now().Add(tlsFirstFallback)) + n, _ := io.ReadFull(c.nc, pre[:]) + c.nc.SetReadDeadline(time.Time{}) + // If we get any data (regardless of possible timeout), we will proceed + // with the TLS handshake. + if n > 0 { + pre = pre[:n] + } else { + // We did not get anything so we will send the INFO protocol. + pre = nil + + // Restore the original info.TLSRequired value if it is + // different that the current value and regenerate infoBytes. + if orgInfoTLSReq != info.TLSRequired { + info.TLSRequired = orgInfoTLSReq + infoBytes = c.generateClientInfoJSON(info) + } + c.sendProtoNow(infoBytes) + // Set the boolean to false for the rest of the function. + tlsFirst = false + // Check closed status again + isClosed = c.isClosed() + } + } // If we have both TLS and non-TLS allowed we need to see which // one the client wants. We'll always allow this for in-process // connections. - if !isClosed && opts.TLSConfig != nil && (inProcess || opts.AllowNonTLS) { + if !isClosed && !tlsFirst && opts.TLSConfig != nil && (inProcess || opts.AllowNonTLS) { pre = make([]byte, 4) c.nc.SetReadDeadline(time.Now().Add(secondsToDuration(opts.TLSTimeout))) n, _ := io.ReadFull(c.nc, pre[:]) @@ -3125,12 +3185,18 @@ func (s *Server) createClientEx(conn net.Conn, inProcess bool) *client { } } - // If connection is marked as closed, bail out. + // Now, send the INFO if it was delayed + if !isClosed && tlsFirst { + c.flags.set(didTLSFirst) + c.sendProtoNow(infoBytes) + // Check closed status + isClosed = c.isClosed() + } + + // Connection could have been closed while sending the INFO proto. if isClosed { c.mu.Unlock() - // Connection could have been closed due to TLS timeout or while trying - // to send the INFO protocol. We need to call closeConnection() to make - // sure that proper cleanup is done. + // We need to call closeConnection() to make sure that proper cleanup is done. c.closeConnection(WriteError) return nil } diff --git a/vendor/github.com/nats-io/nats-server/v2/server/signal_windows.go b/vendor/github.com/nats-io/nats-server/v2/server/signal_windows.go index d55589fe0..b262bc0bd 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/signal_windows.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/signal_windows.go @@ -34,10 +34,15 @@ func (s *Server) handleSignals() { signal.Notify(c, os.Interrupt, syscall.SIGTERM) go func() { - for sig := range c { - s.Debugf("Trapped %q signal", sig) - s.Shutdown() - os.Exit(0) + for { + select { + case sig := <-c: + s.Debugf("Trapped %q signal", sig) + s.Shutdown() + os.Exit(0) + case <-s.quitCh: + return + } } }() } diff --git a/vendor/github.com/nats-io/nats-server/v2/server/stream.go b/vendor/github.com/nats-io/nats-server/v2/server/stream.go index 72ca5f354..ea34bf55e 100644 --- a/vendor/github.com/nats-io/nats-server/v2/server/stream.go +++ b/vendor/github.com/nats-io/nats-server/v2/server/stream.go @@ -160,9 +160,10 @@ type StreamAlternate struct { // ClusterInfo shows information about the underlying set of servers // that make up the stream or consumer. type ClusterInfo struct { - Name string `json:"name,omitempty"` - Leader string `json:"leader,omitempty"` - Replicas []*PeerInfo `json:"replicas,omitempty"` + Name string `json:"name,omitempty"` + RaftGroup string `json:"raft_group,omitempty"` + Leader string `json:"leader,omitempty"` + Replicas []*PeerInfo `json:"replicas,omitempty"` } // PeerInfo shows information about all the peers in the cluster that @@ -2126,6 +2127,14 @@ func (mset *stream) processMirrorMsgs(mirror *sourceInfo, ready *sync.WaitGroup) // Signal the caller that we have captured the above fields. ready.Done() + // Make sure we have valid ipq for msgs. + if msgs == nil { + mset.mu.Lock() + mset.cancelMirrorConsumer() + mset.mu.Unlock() + return + } + t := time.NewTicker(sourceHealthCheckInterval) defer t.Stop() @@ -5266,9 +5275,13 @@ func (mset *stream) Store() StreamStore { // Determines if the new proposed partition is unique amongst all consumers. // Lock should be held. -func (mset *stream) partitionUnique(partitions []string) bool { +func (mset *stream) partitionUnique(name string, partitions []string) bool { for _, partition := range partitions { - for _, o := range mset.consumers { + for n, o := range mset.consumers { + // Skip the consumer being checked. + if n == name { + continue + } if o.subjf == nil { return false } diff --git a/vendor/github.com/nats-io/nats.go/.travis.yml b/vendor/github.com/nats-io/nats.go/.travis.yml index 368797051..1505f773d 100644 --- a/vendor/github.com/nats-io/nats.go/.travis.yml +++ b/vendor/github.com/nats-io/nats.go/.travis.yml @@ -22,7 +22,7 @@ before_script: - golangci-lint run ./jetstream/... script: - go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off -- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi +- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing; fi after_success: - if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi diff --git a/vendor/github.com/nats-io/nats.go/README.md b/vendor/github.com/nats-io/nats.go/README.md index 108db4e35..042733da1 100644 --- a/vendor/github.com/nats-io/nats.go/README.md +++ b/vendor/github.com/nats-io/nats.go/README.md @@ -29,7 +29,7 @@ When using or transitioning to Go modules support: ```bash # Go client latest or explicit version go get github.com/nats-io/nats.go/@latest -go get github.com/nats-io/nats.go/@v1.30.2 +go get github.com/nats-io/nats.go/@v1.31.0 # For latest NATS Server, add /v2 at the end go get github.com/nats-io/nats-server/v2 diff --git a/vendor/github.com/nats-io/nats.go/dependencies.md b/vendor/github.com/nats-io/nats.go/dependencies.md index cc986b277..ec9ab3c6a 100644 --- a/vendor/github.com/nats-io/nats.go/dependencies.md +++ b/vendor/github.com/nats-io/nats.go/dependencies.md @@ -2,12 +2,14 @@ This file lists the dependencies used in this repository. -| Dependency | License | -|-|-| -| Go | BSD 3-Clause "New" or "Revised" License | -| github.com/nats-io/nats.go | Apache License 2.0 | -| github.com/golang/protobuf v1.4.2 | BSD 3-Clause "New" or "Revised" License | -| github.com/nats-io/nats-server/v2 v2.1.8-0.20201115145023-f61fa8529a0f | Apache License 2.0 | -| github.com/nats-io/nkeys v0.2.0 | Apache License 2.0 | -| github.com/nats-io/nuid v1.0.1 | Apache License 2.0 | -| google.golang.org/protobuf v1.23.0 | BSD 3-Clause License | +| Dependency | License | +|-----------------------------------|--------------| +| Go | BSD 3-Clause | +| github.com/golang/protobuf/proto | BSD-3-Clause | +| github.com/klauspost/compress | BSD-3-Clause | +| github.com/nats-io/nats-server/v2 | Apache-2.0 | +| github.com/nats-io/nkeys | Apache-2.0 | +| github.com/nats-io/nuid | Apache-2.0 | +| go.uber.org/goleak | MIT | +| golang.org/x/text | BSD-3-Clause | +| google.golang.org/protobuf | BSD-3-Clause | diff --git a/vendor/github.com/nats-io/nats.go/js.go b/vendor/github.com/nats-io/nats.go/js.go index 7fdb0131c..444278e0f 100644 --- a/vendor/github.com/nats-io/nats.go/js.go +++ b/vendor/github.com/nats-io/nats.go/js.go @@ -3223,6 +3223,9 @@ func (js *js) getConsumerInfoContext(ctx context.Context, stream, consumer strin } return nil, info.Error } + if info.Error == nil && info.ConsumerInfo == nil { + return nil, ErrConsumerNotFound + } return info.ConsumerInfo, nil } diff --git a/vendor/github.com/nats-io/nats.go/nats.go b/vendor/github.com/nats-io/nats.go/nats.go index 82b79730a..da13692fd 100644 --- a/vendor/github.com/nats-io/nats.go/nats.go +++ b/vendor/github.com/nats-io/nats.go/nats.go @@ -47,7 +47,7 @@ import ( // Default Constants const ( - Version = "1.30.2" + Version = "1.31.0" DefaultURL = "nats://127.0.0.1:4222" DefaultPort = 4222 DefaultMaxReconnect = 60 @@ -311,6 +311,13 @@ type Options struct { // TLSCertCB is used to fetch and return custom tls certificate. TLSCertCB TLSCertHandler + // TLSHandshakeFirst is used to instruct the library perform + // the TLS handshake right after the connect and before receiving + // the INFO protocol from the server. If this option is enabled + // but the server is not configured to perform the TLS handshake + // first, the connection will fail. + TLSHandshakeFirst bool + // RootCAsCB is used to fetch and return a set of root certificate // authorities that clients use when verifying server certificates. RootCAsCB RootCAsHandler @@ -1315,6 +1322,17 @@ func SkipHostLookup() Option { } } +// TLSHandshakeFirst is an Option to perform the TLS handshake first, that is +// before receiving the INFO protocol. This requires the server to also be +// configured with such option, otherwise the connection will fail. +func TLSHandshakeFirst() Option { + return func(o *Options) error { + o.TLSHandshakeFirst = true + o.Secure = true + return nil + } +} + // Handler processing // SetDisconnectHandler will set the disconnect event handler. @@ -1481,6 +1499,12 @@ func (o Options) Connect() (*Conn, error) { } } + // If the TLSHandshakeFirst option is specified, make sure that + // the Secure boolean is true. + if nc.Opts.TLSHandshakeFirst { + nc.Opts.Secure = true + } + if err := nc.setupServerPool(); err != nil { return nil, err } @@ -2235,6 +2259,14 @@ func (nc *Conn) processConnectInit() error { // Set our status to connecting. nc.changeConnStatus(CONNECTING) + // If we need to have a TLS connection and want the TLS handshake to occur + // first, do it now. + if nc.Opts.Secure && nc.Opts.TLSHandshakeFirst { + if err := nc.makeTLSConn(); err != nil { + return err + } + } + // Process the INFO protocol received from the server err := nc.processExpectedInfo() if err != nil { @@ -2351,8 +2383,13 @@ func (nc *Conn) checkForSecure() error { o.Secure = true } - // Need to rewrap with bufio if o.Secure { + // If TLS handshake first is true, we have already done + // the handshake, so we are done here. + if o.TLSHandshakeFirst { + return nil + } + // Need to rewrap with bufio if err := nc.makeTLSConn(); err != nil { return err } @@ -3040,28 +3077,6 @@ func (nc *Conn) waitForMsgs(s *Subscription) { // Return what is to be used. If we return nil the message will be dropped. type msgFilter func(m *Msg) *Msg -func (nc *Conn) addMsgFilter(subject string, filter msgFilter) { - nc.subsMu.Lock() - defer nc.subsMu.Unlock() - - if nc.filters == nil { - nc.filters = make(map[string]msgFilter) - } - nc.filters[subject] = filter -} - -func (nc *Conn) removeMsgFilter(subject string) { - nc.subsMu.Lock() - defer nc.subsMu.Unlock() - - if nc.filters != nil { - delete(nc.filters, subject) - if len(nc.filters) == 0 { - nc.filters = nil - } - } -} - // processMsg is called by parse and will place the msg on the // appropriate channel/pending queue for processing. If the channel is full, // or the pending queue is over the pending limits, the connection is diff --git a/vendor/github.com/nats-io/nats.go/testing_internal.go b/vendor/github.com/nats-io/nats.go/testing_internal.go new file mode 100644 index 000000000..18397026a --- /dev/null +++ b/vendor/github.com/nats-io/nats.go/testing_internal.go @@ -0,0 +1,59 @@ +// Copyright 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 +// +// 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. + +//go:build internal_testing +// +build internal_testing + +// Functions in this file are only available when building nats.go with the +// internal_testing build tag. They are used by the nats.go test suite. +package nats + +// AddMsgFilter adds a message filter for the given subject +// to the connection. The filter will be called for each +// message received on the subject. If the filter returns +// nil, the message will be dropped. +func (nc *Conn) AddMsgFilter(subject string, filter msgFilter) { + nc.subsMu.Lock() + defer nc.subsMu.Unlock() + + if nc.filters == nil { + nc.filters = make(map[string]msgFilter) + } + nc.filters[subject] = filter +} + +// RemoveMsgFilter removes a message filter for the given subject. +func (nc *Conn) RemoveMsgFilter(subject string) { + nc.subsMu.Lock() + defer nc.subsMu.Unlock() + + if nc.filters != nil { + delete(nc.filters, subject) + if len(nc.filters) == 0 { + nc.filters = nil + } + } +} + +// IsJSControlMessage returns true if the message is a JetStream control message. +func IsJSControlMessage(msg *Msg) (bool, int) { + return isJSControlMessage(msg) +} + +// CloseTCPConn closes the underlying TCP connection. +// It can be used to simulate a disconnect. +func (nc *Conn) CloseTCPConn() { + nc.mu.Lock() + defer nc.mu.Unlock() + nc.conn.Close() +} diff --git a/vendor/github.com/nats-io/nkeys/nkeys.go b/vendor/github.com/nats-io/nkeys/nkeys.go index b83a99d4f..0db0f0c1f 100644 --- a/vendor/github.com/nats-io/nkeys/nkeys.go +++ b/vendor/github.com/nats-io/nkeys/nkeys.go @@ -19,7 +19,7 @@ package nkeys import "io" // Version is our current version -const Version = "0.4.5" +const Version = "0.4.6" // KeyPair provides the central interface to nkeys. type KeyPair interface { diff --git a/vendor/github.com/nats-io/nkeys/xkeys.go b/vendor/github.com/nats-io/nkeys/xkeys.go index 54e50b9de..78f8b99e1 100644 --- a/vendor/github.com/nats-io/nkeys/xkeys.go +++ b/vendor/github.com/nats-io/nkeys/xkeys.go @@ -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 @@ -38,12 +38,13 @@ type ckp struct { seed [curveKeyLen]byte // Private raw key. } -// CreateUser will create a User typed KeyPair. +// CreateCurveKeys will create a Curve typed KeyPair. func CreateCurveKeys() (KeyPair, error) { return CreateCurveKeysWithRand(rand.Reader) } -// CreateUser will create a User typed KeyPair with specified rand source. +// CreateCurveKeysWithRand will create a Curve typed KeyPair +// with specified rand source. func CreateCurveKeysWithRand(rr io.Reader) (KeyPair, error) { var kp ckp _, err := io.ReadFull(rr, kp.seed[:]) @@ -85,7 +86,7 @@ func (pair *ckp) PrivateKey() ([]byte, error) { return Encode(PrefixBytePrivate, pair.seed[:]) } -func decodePubCurveKey(src string, dest [curveKeyLen]byte) error { +func decodePubCurveKey(src string, dest []byte) error { var raw [curveDecodeLen]byte // should always be 35 n, err := b32Enc.Decode(raw[:], []byte(src)) if err != nil { @@ -112,7 +113,7 @@ func decodePubCurveKey(src string, dest [curveKeyLen]byte) error { } // Copy over, ignore prefix byte. - copy(dest[:], raw[1:end]) + copy(dest, raw[1:end]) return nil } @@ -134,7 +135,7 @@ func (pair *ckp) SealWithRand(input []byte, recipient string, rr io.Reader) ([]b err error ) - if err = decodePubCurveKey(recipient, rpub); err != nil { + if err = decodePubCurveKey(recipient, rpub[:]); err != nil { return nil, ErrInvalidRecipient } if _, err := io.ReadFull(rr, nonce[:]); err != nil { @@ -159,7 +160,7 @@ func (pair *ckp) Open(input []byte, sender string) ([]byte, error) { } copy(nonce[:], input[vlen:vlen+curveNonceLen]) - if err = decodePubCurveKey(sender, spub); err != nil { + if err = decodePubCurveKey(sender, spub[:]); err != nil { return nil, ErrInvalidSender } diff --git a/vendor/modules.txt b/vendor/modules.txt index 2ac579227..d767b9241 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1210,7 +1210,7 @@ github.com/justinas/alice # github.com/kevinburke/ssh_config v1.2.0 ## explicit github.com/kevinburke/ssh_config -# github.com/klauspost/compress v1.17.0 +# github.com/klauspost/compress v1.17.2 ## explicit; go 1.18 github.com/klauspost/compress/flate github.com/klauspost/compress/s2 @@ -1367,7 +1367,7 @@ github.com/mschoch/smat # github.com/nats-io/jwt/v2 v2.5.2 ## explicit; go 1.18 github.com/nats-io/jwt/v2 -# github.com/nats-io/nats-server/v2 v2.10.2 +# github.com/nats-io/nats-server/v2 v2.10.4 ## explicit; go 1.20 github.com/nats-io/nats-server/v2/conf github.com/nats-io/nats-server/v2/internal/ldap @@ -1378,13 +1378,13 @@ github.com/nats-io/nats-server/v2/server/certidp github.com/nats-io/nats-server/v2/server/certstore github.com/nats-io/nats-server/v2/server/pse github.com/nats-io/nats-server/v2/server/sysmem -# github.com/nats-io/nats.go v1.30.2 +# github.com/nats-io/nats.go v1.31.0 ## explicit; go 1.20 github.com/nats-io/nats.go github.com/nats-io/nats.go/encoders/builtin github.com/nats-io/nats.go/internal/parser github.com/nats-io/nats.go/util -# github.com/nats-io/nkeys v0.4.5 +# github.com/nats-io/nkeys v0.4.6 ## explicit; go 1.19 github.com/nats-io/nkeys # github.com/nats-io/nuid v1.0.1