Repository: nats-io/nats-server
Pull request #1946 (2021-02-28 19:03:49)

upgrade nightlies to go 1.16
Signed-off-by: R.I.Pienaar <rip@devco.net>

/cc @nats-io/core



Patch:
diff --git a/.github/workflows/nightly.yaml b/.github/workflows/nightly.yaml
index 83011b90..d76d2b12 100644
--- a/.github/workflows/nightly.yaml
+++ b/.github/workflows/nightly.yaml
@@ -23,7 +23,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.15
+          go-version: 1.16
 
       - name: goreleaser
         uses: goreleaser/goreleaser-action@v2
diff --git a/docker/Dockerfile.nightly b/docker/Dockerfile.nightly
index 333d848f..fc86f64f 100644
--- a/docker/Dockerfile.nightly
+++ b/docker/Dockerfile.nightly
@@ -1,4 +1,4 @@
-FROM golang:1.15-alpine AS builder
+FROM golang:1.16-alpine AS builder
 
 ARG VERSION="nightly"
 
diff --git a/server/const.go b/server/const.go
index 4445eee4..4f8b60ff 100644
--- a/server/const.go
+++ b/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.96"
+	VERSION = "2.2.0-beta.94"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index c397c7f3..f1a335b1 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1552,10 +1552,8 @@ func nextReqFromMsg(msg []byte) (time.Time, int, bool, error) {
 		if err := json.Unmarshal(req, &cr); err != nil {
 			return time.Time{}, -1, false, err
 		}
-		if cr.Expires == time.Duration(0) {
-			return time.Time{}, cr.Batch, cr.NoWait, nil
-		}
-		return time.Now().Add(cr.Expires), cr.Batch, cr.NoWait, nil
+		return cr.Expires, cr.Batch, cr.NoWait, nil
+
 	default:
 		if n, err := strconv.Atoi(string(req)); err == nil {
 			return time.Time{}, n, false, nil
diff --git a/server/jetstream.go b/server/jetstream.go
index 89cd2adf..fb375dc8 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -249,7 +249,7 @@ func (s *Server) setJetStreamDisabled() {
 	js.mu.Unlock()
 }
 
-func (s *Server) handleOutOfSpace(stream string) {
+func (s *Server) handleOutOfSpace() {
 	if s.JetStreamEnabled() {
 		s.Errorf("JetStream out of space, will be DISABLED")
 		go s.DisableJetStream()
@@ -260,10 +260,7 @@ func (s *Server) handleOutOfSpace(stream string) {
 				ID:   nuid.Next(),
 				Time: time.Now().UTC(),
 			},
-			Server:   s.Name(),
-			ServerID: s.ID(),
-			Stream:   stream,
-			Cluster:  s.cachedClusterName(),
+			Server: s.Name(),
 		}
 		s.publishAdvisory(nil, JSAdvisoryServerOutOfStorage, adv)
 	}
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index bc0ea0c8..88731617 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -505,9 +505,9 @@ const JSApiConsumerListResponseType = "io.nats.jetstream.api.v1.consumer_list_re
 
 // JSApiConsumerGetNextRequest is for getting next messages for pull based consumers.
 type JSApiConsumerGetNextRequest struct {
-	Expires time.Duration `json:"expires,omitempty"`
-	Batch   int           `json:"batch,omitempty"`
-	NoWait  bool          `json:"no_wait,omitempty"`
+	Expires time.Time `json:"expires,omitempty"`
+	Batch   int       `json:"batch,omitempty"`
+	NoWait  bool      `json:"no_wait,omitempty"`
 }
 
 // JSApiStreamTemplateCreateResponse for creating templates.
@@ -1050,7 +1050,7 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, subject, re
 			return
 		}
 
-		// We do not require other stream to exist anymore, but if we can see it check payloads.
+		// Now make sure the other stream exists.
 		var exists bool
 		var maxMsgSize int32
 
@@ -1064,7 +1064,12 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, subject, re
 			maxMsgSize = mset.cfg.MaxMsgSize
 			exists = true
 		}
-		if exists && cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
+		if !exists {
+			resp.Error = &ApiError{Code: 400, Description: "stream mirror source must exist"}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		if cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
 			resp.Error = &ApiError{Code: 400, Description: "stream mirror must have max message size >= source"}
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 44267335..cd4cb954 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1420,7 +1420,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 						s.Debugf("Got error processing JetStream msg: %v", err)
 					}
 					if isOutOfSpaceErr(err) {
-						s.handleOutOfSpace(mset.name())
+						s.handleOutOfSpace()
 						return err
 					}
 				}
@@ -3744,7 +3744,6 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	canRespond := !mset.cfg.NoAck && len(reply) > 0
 	s, jsa, st, rf, sendq := mset.srv, mset.jsa, mset.cfg.Storage, mset.cfg.Replicas, mset.sendq
 	maxMsgSize := int(mset.cfg.MaxMsgSize)
-	msetName := mset.cfg.Name
 	mset.mu.RUnlock()
 
 	// Check here pre-emptively if we have exceeded our account limits.
@@ -3817,7 +3816,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	}
 
 	if err != nil && isOutOfSpaceErr(err) {
-		s.handleOutOfSpace(msetName)
+		s.handleOutOfSpace()
 	}
 
 	return err
@@ -3884,7 +3883,6 @@ func (mset *stream) processSnapshot(snap *streamSnapshot) {
 	state := mset.store.State()
 	sreq := mset.calculateSyncRequest(&state, snap)
 	s, subject, n := mset.srv, mset.sa.Sync, mset.node
-	msetName := mset.cfg.Name
 	mset.mu.Unlock()
 
 	// Just return if up to date..
@@ -3990,7 +3988,7 @@ RETRY:
 					return
 				}
 			} else if isOutOfSpaceErr(err) {
-				s.handleOutOfSpace(msetName)
+				s.handleOutOfSpace()
 				return
 			} else {
 				goto RETRY
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 45ebb1c2..723db078 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -210,8 +210,5 @@ const JSServerOutOfStorageAdvisoryType = "io.nats.jetstream.advisory.v1.server_o
 // JSServerOutOfSpaceAdvisory indicates that a stream has lost quorum and is stalled.
 type JSServerOutOfSpaceAdvisory struct {
 	TypedEvent
-	Server   string `json:"server"`
-	ServerID string `json:"server_id"`
-	Stream   string `json:"stream,omitempty"`
-	Cluster  string `json:"cluster"`
+	Server string `json:"server"`
 }
diff --git a/server/raft.go b/server/raft.go
index 043a8ab0..c5061c14 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -2661,7 +2661,7 @@ func (n *raft) setWriteErrLocked(err error) {
 
 	// For now since this can be happening all under the covers, we will call up and disable JetStream.
 	n.Unlock()
-	n.s.handleOutOfSpace("")
+	n.s.handleOutOfSpace()
 	n.Lock()
 }
 
diff --git a/server/stream.go b/server/stream.go
index dc2b0417..f18266c0 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -214,7 +214,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	// This can happen on startup with restored state where on meta replay we still do not have
 	// the assignment. Running in single server mode this always returns true.
 	if !jsa.streamAssigned(config.Name) {
-		s.Debugf("Stream '%s > %s' does not seem to be assigned to this server", a.Name, config.Name)
+		s.Debugf("Stream %q does not seem to be assigned to this server", config.Name)
 	}
 
 	// Sensible defaults.
@@ -1015,15 +1015,6 @@ func (mset *stream) cancelMirrorConsumer() {
 	}
 }
 
-func (mset *stream) retryMirrorConsumer() {
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
-	mset.srv.Debugf("Retrying mirror consumer for '%s > %s'", mset.acc.Name, mset.cfg.Name)
-	if mset.mirror != nil && mset.mirror.sub == nil {
-		mset.setupMirrorConsumer()
-	}
-}
-
 func (mset *stream) resetMirrorConsumer() error {
 	mset.mu.Lock()
 	defer mset.mu.Unlock()
@@ -1115,8 +1106,6 @@ func (mset *stream) setupMirrorConsumer() error {
 			if ccr.Error != nil {
 				mset.cancelMirrorConsumer()
 				shouldRetry = false
-				// We will retry every 5 seconds or so
-				time.AfterFunc(5*time.Second, mset.retryMirrorConsumer)
 			}
 			mset.setMirrorErr(ccr.Error)
 		case <-time.After(2 * time.Second):
@@ -1143,32 +1132,6 @@ func (mset *stream) streamSource(sname string) *StreamSource {
 	return nil
 }
 
-func (mset *stream) retrySourceConsumer(sname string) {
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
-	if mset.client == nil {
-		return
-	}
-	s := mset.srv
-	s.Debugf("Retrying source consumer for '%s > %s'", mset.acc.Name, mset.cfg.Name)
-	si := mset.sources[sname]
-	// No longer configured or still active.
-	if si == nil || si.sub != nil {
-		return
-	}
-	mset.setStartingSequenceForSource(sname)
-	mset.setSourceConsumer(sname, si.sseq+1)
-}
-
-// Locl should be held.
-func (mset *stream) cancelSourceConsumer(sname string) {
-	if si := mset.sources[sname]; si != nil && si.sub != nil {
-		mset.unsubscribe(si.sub)
-		si.sub = nil
-		si.sseq, si.dseq = 0, 0
-	}
-}
-
 // Lock should be held.
 func (mset *stream) setSourceConsumer(sname string, seq uint64) {
 	si := mset.sources[sname]
@@ -1250,14 +1213,10 @@ func (mset *stream) setSourceConsumer(sname string, seq uint64) {
 		case ccr := <-respCh:
 			mset.mu.Lock()
 			if si := mset.sources[sname]; si != nil {
-				si.err = nil
 				if ccr.Error != nil {
 					mset.srv.Warnf("JetStream error response for create source consumer: %+v", ccr.Error)
 					si.err = ccr.Error
 					shouldRetry = false
-					// We will retry every 5 seconds or so
-					mset.cancelSourceConsumer(sname)
-					time.AfterFunc(5*time.Second, func() { mset.retrySourceConsumer(sname) })
 				}
 			}
 			mset.mu.Unlock()
@@ -1307,10 +1266,6 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, sub *subscription, c
 
 	// Hold onto the origin reply which has all the metadata.
 	hdr, msg := c.msgParts(rmsg)
-	// If we are daisy chained here make sure to remove the original one.
-	if len(hdr) > 0 {
-		hdr = removeHeaderIfPresent(hdr, JSStreamSource)
-	}
 	hdr = genHeader(hdr, JSStreamSource, reply)
 
 	var err error
@@ -1346,37 +1301,6 @@ func streamAndSeq(subject string) (string, uint64) {
 	return tokens[2], uint64(parseAckReplyNum(tokens[5]))
 }
 
-// Lock should be held.
-func (mset *stream) setStartingSequenceForSource(sname string) {
-	si := mset.sources[sname]
-	if si == nil {
-		return
-	}
-
-	state := mset.store.State()
-	if state.Msgs == 0 {
-		si.sseq, si.dseq = 0, 1
-		return
-	}
-
-	for seq := state.LastSeq; seq >= state.FirstSeq; seq-- {
-		_, hdr, _, _, err := mset.store.LoadMsg(seq)
-		if err != nil || len(hdr) == 0 {
-			continue
-		}
-		reply := getHeader(JSStreamSource, hdr)
-		if len(reply) == 0 {
-			continue
-		}
-		name, sseq := streamAndSeq(string(reply))
-		if name == sname {
-			si.sseq = sseq
-			si.dseq = 1
-			return
-		}
-	}
-}
-
 // Lock should be held.
 // This will do a reverse scan on startup or leader election
 // searching for the starting sequence number.


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 8c95d294..fefc38f1 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -3596,16 +3596,6 @@ func TestJetStreamClusterSuperClusterCrossClusterConsumerInterest(t *testing.T)
 	checkSubsPending(t, sub, 2)
 }
 
-func TestJetStreamNextReqFromMsg(t *testing.T) {
-	bef := time.Now()
-	expires, _, _, err := nextReqFromMsg([]byte(`{"expires":5000000000}`)) // nanoseconds
-	require_NoError(t, err)
-	now := time.Now()
-	if expires.Before(bef.Add(5*time.Second)) || expires.After(now.Add(5*time.Second)) {
-		t.Fatal("Expires out of expected range")
-	}
-}
-
 func TestJetStreamClusterSuperClusterPeerReassign(t *testing.T) {
 	sc := createJetStreamSuperCluster(t, 3, 3)
 	defer sc.shutdown()
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index ddf410ed..9a3c1281 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -1784,7 +1784,7 @@ func TestJetStreamWorkQueueRequest(t *testing.T) {
 			// Now do expiration
 			req.Batch = 1
 			req.NoWait = false
-			req.Expires = 10 * time.Millisecond
+			req.Expires = time.Now().Add(10 * time.Millisecond)
 			jreq, _ = json.Marshal(req)
 
 			nc.PublishRequest(getSubj, reply, jreq)
@@ -10441,8 +10441,11 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	// Clear subjects.
 	cfg.Subjects = nil
+	scr = createStream(cfg)
+	if scr.Error == nil || !strings.Contains(scr.Error.Description, "stream mirror source must exist") {
+		t.Fatalf("Expected error, got %+v", scr.Error)
+	}
 
-	// Source
 	scfg := &StreamConfig{
 		Name:     "S1",
 		Storage:  FileStorage,