Repository: nats-io/nats-server
Pull request #3132 (2022-05-18 15:11:44)

With use cases bringing us more data I wanted to suggest these changes.
With inlining election timeout updates we double the lock contention and most likely introduced head of line issues for routes under heavy load.
Also slowing down heartbeats with so many assets being deployed in our user ecosystem, also moved the normal follower to candidate timing further out, similar to the lost quorum.
Note that the happy path transfer will still be very quick.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/const.go b/server/const.go
index a0d12087..3074ef0b 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.8.3-beta.2"
+	VERSION = "2.8.3-beta.1"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index 7ac787e3..357e1e97 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -68,9 +68,8 @@ type ConsumerConfig struct {
 	HeadersOnly     bool            `json:"headers_only,omitempty"`
 
 	// Pull based options.
-	MaxRequestBatch    int           `json:"max_batch,omitempty"`
-	MaxRequestExpires  time.Duration `json:"max_expires,omitempty"`
-	MaxRequestMaxBytes int           `json:"max_bytes,omitempty"`
+	MaxRequestBatch   int           `json:"max_batch,omitempty"`
+	MaxRequestExpires time.Duration `json:"max_expires,omitempty"`
 
 	// Push based consumers.
 	DeliverSubject string `json:"deliver_subject,omitempty"`
@@ -79,11 +78,6 @@ type ConsumerConfig struct {
 	// Ephemeral inactivity threshold.
 	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
 
-	// Generally inherited by parent stream and other markers, now can be configured directly.
-	Replicas int `json:"num_replicas"`
-	// Force memory storage.
-	MemoryStorage bool `json:"mem_storage,omitempty"`
-
 	// Don't add to general clients.
 	Direct bool `json:"direct,omitempty"`
 }
@@ -356,11 +350,6 @@ func checkConsumerCfg(
 	isRecovering bool,
 ) *ApiError {
 
-	// Check if replicas is defined but exceeds parent stream.
-	if config.Replicas > 0 && config.Replicas > cfg.Replicas {
-		return NewJSConsumerReplicasExceedsStreamError()
-	}
-
 	// Check if we have a BackOff defined that MaxDeliver is within range etc.
 	if lbo := len(config.BackOff); lbo > 0 && config.MaxDeliver <= lbo {
 		return NewJSConsumerMaxDeliverBackoffError()
@@ -2307,36 +2296,36 @@ func (o *consumer) needAck(sseq uint64) bool {
 }
 
 // Helper for the next message requests.
-func nextReqFromMsg(msg []byte) (time.Time, int, int, bool, time.Duration, time.Time, error) {
+func nextReqFromMsg(msg []byte) (time.Time, int, bool, time.Duration, time.Time, error) {
 	req := bytes.TrimSpace(msg)
 
 	switch {
 	case len(req) == 0:
-		return time.Time{}, 1, 0, false, 0, time.Time{}, nil
+		return time.Time{}, 1, false, 0, time.Time{}, nil
 
 	case req[0] == '{':
 		var cr JSApiConsumerGetNextRequest
 		if err := json.Unmarshal(req, &cr); err != nil {
-			return time.Time{}, -1, 0, false, 0, time.Time{}, err
+			return time.Time{}, -1, false, 0, time.Time{}, err
 		}
 		var hbt time.Time
 		if cr.Heartbeat > 0 {
 			if cr.Heartbeat*2 > cr.Expires {
-				return time.Time{}, 1, 0, false, 0, time.Time{}, errors.New("heartbeat value too large")
+				return time.Time{}, 1, false, 0, time.Time{}, errors.New("heartbeat value too large")
 			}
 			hbt = time.Now().Add(cr.Heartbeat)
 		}
 		if cr.Expires == time.Duration(0) {
-			return time.Time{}, cr.Batch, cr.MaxBytes, cr.NoWait, cr.Heartbeat, hbt, nil
+			return time.Time{}, cr.Batch, cr.NoWait, cr.Heartbeat, hbt, nil
 		}
-		return time.Now().Add(cr.Expires), cr.Batch, cr.MaxBytes, cr.NoWait, cr.Heartbeat, hbt, nil
+		return time.Now().Add(cr.Expires), cr.Batch, cr.NoWait, cr.Heartbeat, hbt, nil
 	default:
 		if n, err := strconv.Atoi(string(req)); err == nil {
-			return time.Time{}, n, 0, false, 0, time.Time{}, nil
+			return time.Time{}, n, false, 0, time.Time{}, nil
 		}
 	}
 
-	return time.Time{}, 1, 0, false, 0, time.Time{}, nil
+	return time.Time{}, 1, false, 0, time.Time{}, nil
 }
 
 // Represents a request that is on the internal waiting queue
@@ -2346,7 +2335,6 @@ type waitingRequest struct {
 	reply    string
 	n        int // For batching
 	d        int
-	b        int // For max bytes tracking.
 	expires  time.Time
 	received time.Time
 	hb       time.Duration
@@ -2513,39 +2501,12 @@ func (o *consumer) pendingRequestReplies() []string {
 // Return next waiting request. This will check for expirations but not noWait or interest.
 // That will be handled by processWaiting.
 // Lock should be held.
-func (o *consumer) nextWaiting(sz int) *waitingRequest {
+func (o *consumer) nextWaiting() *waitingRequest {
 	if o.waiting == nil || o.waiting.isEmpty() {
 		return nil
 	}
 	for wr := o.waiting.peek(); !o.waiting.isEmpty(); wr = o.waiting.peek() {
-		if wr == nil {
-			break
-		}
-		// Check if we have max bytes set.
-		if wr.b > 0 {
-			if sz <= wr.b {
-				wr.b -= sz
-				// If we are right now at zero, set batch to 1 to deliver this one but stop after.
-				if wr.b == 0 {
-					wr.n = 1
-				}
-			} else {
-				// If we have not delivered anything to the requestor let them know.
-				if wr.d == 0 {
-					hdr := []byte("NATS/1.0 408 Message Size Exceeds MaxBytes\r\n\r\n")
-					o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
-				}
-				// Remove the current one, no longer valid due to max bytes limit.
-				o.waiting.removeCurrent()
-				if o.node != nil {
-					o.removeClusterPendingRequest(wr.reply)
-				}
-				wr.recycle()
-				continue
-			}
-		}
-
-		if wr.expires.IsZero() || time.Now().Before(wr.expires) {
+		if wr == nil || wr.expires.IsZero() || time.Now().Before(wr.expires) {
 			rr := wr.acc.sl.Match(wr.interest)
 			if len(rr.psubs)+len(rr.qsubs) > 0 {
 				return o.waiting.pop()
@@ -2600,7 +2561,7 @@ func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
 	}
 
 	// Check payload here to see if they sent in batch size or a formal request.
-	expires, batchSize, maxBytes, noWait, hb, hbt, err := nextReqFromMsg(msg)
+	expires, batchSize, noWait, hb, hbt, err := nextReqFromMsg(msg)
 	if err != nil {
 		sendErr(400, fmt.Sprintf("Bad Request - %v", err))
 		return
@@ -2617,11 +2578,6 @@ func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
 		return
 	}
 
-	if maxBytes > 0 && o.cfg.MaxRequestMaxBytes > 0 && maxBytes > o.cfg.MaxRequestMaxBytes {
-		sendErr(409, fmt.Sprintf("Exceeded MaxRequestMaxBytes of %v", o.cfg.MaxRequestMaxBytes))
-		return
-	}
-
 	// If we have the max number of requests already pending try to expire.
 	if o.waiting.isFull() {
 		// Try to expire some of the requests.
@@ -2651,10 +2607,9 @@ func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
 	// If we receive this request though an account export, we need to track that interest subject and account.
 	acc, interest := trackDownAccountAndInterest(o.acc, reply)
 
-	// Create a waiting request.
+	// In case we have to queue up this request.
 	wr := wrPool.Get().(*waitingRequest)
 	wr.acc, wr.interest, wr.reply, wr.n, wr.d, wr.noWait, wr.expires, wr.hb, wr.hbt = acc, interest, reply, batchSize, 0, noWait, expires, hb, hbt
-	wr.b = maxBytes
 	wr.received = time.Now()
 
 	if err := o.waiting.add(wr); err != nil {
@@ -2991,7 +2946,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 
 		if o.isPushMode() {
 			dsubj = o.dsubj
-		} else if wr := o.nextWaiting(len(pmsg.hdr) + len(pmsg.msg)); wr != nil {
+		} else if wr := o.nextWaiting(); wr != nil {
 			dsubj = wr.reply
 			if done := wr.recycleIfDone(); done && o.node != nil {
 				o.removeClusterPendingRequest(dsubj)
diff --git a/server/errors.json b/server/errors.json
index 1192377b..bf2ff9b6 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1238,15 +1238,5 @@
     "help": "",
     "url": "",
     "deprecates": ""
-  },
-  {
-    "constant": "JSConsumerReplicasExceedsStream",
-    "code": 400,
-    "error_code": 10126,
-    "description": "consumer config replica count exceeds parent stream",
-    "comment": "",
-    "help": "",
-    "url": "",
-    "deprecates": ""
   }
 ]
\ No newline at end of file
diff --git a/server/filestore.go b/server/filestore.go
index 808a792c..0bcaf2d7 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -856,10 +856,7 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 		// This is an old erased message, or a new one that we can track.
 		if seq == 0 || seq&ebit != 0 || seq < mb.first.seq {
 			seq = seq &^ ebit
-			// Only add to dmap if past recorded first seq and non-zero.
-			if seq != 0 && seq >= mb.first.seq {
-				addToDmap(seq)
-			}
+			addToDmap(seq)
 			index += rl
 			mb.last.seq = seq
 			mb.last.ts = ts
@@ -1097,14 +1094,10 @@ func (fs *fileStore) expireMsgsOnRecover() {
 						mb.dmap = nil
 					}
 				}
-				// Keep this update just in case since we are removing dmap entries.
-				mb.first.seq = seq
 				continue
 			}
 			// Break on other errors.
 			if err != nil || sm == nil {
-				// Keep this update just in case since we could have removed dmap entries.
-				mb.first.seq = seq
 				break
 			}
 
@@ -1119,13 +1112,11 @@ func (fs *fileStore) expireMsgsOnRecover() {
 			}
 
 			// Delete the message here.
-			if mb.msgs > 0 {
-				sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-				mb.bytes -= sz
-				bytes += sz
-				mb.msgs--
-				purged++
-			}
+			sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
+			mb.bytes -= sz
+			bytes += sz
+			mb.msgs--
+			purged++
 			// Update fss
 			fs.removePerSubject(sm.subj)
 			mb.removeSeqPerSubject(sm.subj, seq, &smv)
@@ -2007,7 +1998,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 
 	mb.mu.Lock()
 
-	// See if the sequence number is still relevant.
+	// See if the sequence numbers is still relevant.
 	if seq < mb.first.seq {
 		mb.mu.Unlock()
 		fsUnlock()
@@ -2179,8 +2170,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 // writing new messages. We will silently bail on any issues with the underlying block and let someone else detect.
 // Write lock needs to be held.
 func (mb *msgBlock) compact() {
-	wasLoaded := mb.cacheAlreadyLoaded()
-	if !wasLoaded {
+	if mb.cacheNotLoaded() {
 		if err := mb.loadMsgsWithLock(); err != nil {
 			return
 		}
@@ -2262,12 +2252,11 @@ func (mb *msgBlock) compact() {
 
 	// We will write to a new file and mv/rename it in case of failure.
 	mfn := filepath.Join(filepath.Join(mb.fs.fcfg.StoreDir, msgDir), fmt.Sprintf(newScan, mb.index))
+	defer os.Remove(mfn)
 	if err := ioutil.WriteFile(mfn, nbuf, defaultFilePerms); err != nil {
-		os.Remove(mfn)
 		return
 	}
 	if err := os.Rename(mfn, mb.mfn); err != nil {
-		os.Remove(mfn)
 		return
 	}
 
@@ -2276,11 +2265,7 @@ func (mb *msgBlock) compact() {
 	mb.removeIndexFileLocked()
 	mb.deleteDmap()
 	mb.rebuildStateLocked()
-
-	// If we entered with the msgs loaded make sure to reload them.
-	if wasLoaded {
-		mb.loadMsgsWithLock()
-	}
+	mb.loadMsgsWithLock()
 }
 
 // Nil out our dmap.
@@ -2500,34 +2485,23 @@ func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 	var purged, bytes uint64
 
 	mb.mu.Lock()
-
 	checkDmap := len(mb.dmap) > 0
-	var smv StoreMsg
-
 	for seq := mb.last.seq; seq > sm.seq; seq-- {
 		if checkDmap {
 			if _, ok := mb.dmap[seq]; ok {
 				// Delete and skip to next.
 				delete(mb.dmap, seq)
-				if len(mb.dmap) == 0 {
-					mb.dmap = nil
-					checkDmap = false
-				}
 				continue
 			}
 		}
 		// We should have a valid msg to calculate removal stats.
-		if m, err := mb.cacheLookup(seq, &smv); err == nil {
-			if mb.msgs > 0 {
-				rl := fileStoreMsgSize(m.subj, m.hdr, m.msg)
-				mb.msgs--
-				mb.bytes -= rl
-				mb.rbytes -= rl
-				// For return accounting.
-				purged++
-				bytes += uint64(rl)
-			}
+		_, rl, _, err := mb.slotInfo(int(seq - mb.cache.fseq))
+		if err != nil {
+			mb.mu.Unlock()
+			return 0, 0, err
 		}
+		purged++
+		bytes += uint64(rl)
 	}
 
 	// Truncate our msgs and close file.
@@ -2543,6 +2517,11 @@ func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 		return 0, 0, fmt.Errorf("failed to truncate msg block %d, file not open", mb.index)
 	}
 
+	// Do local mb stat updates.
+	mb.msgs -= purged
+	mb.bytes -= bytes
+	mb.rbytes -= bytes
+
 	// Update our last msg.
 	mb.last.seq = sm.seq
 	mb.last.ts = sm.ts
@@ -3624,12 +3603,6 @@ func (mb *msgBlock) cacheLookup(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 	if err != nil || fsm == nil {
 		return nil, err
 	}
-
-	// Deleted messages that are decoded return a 0 for seqeunce.
-	if fsm.seq == 0 {
-		return nil, errDeletedMsg
-	}
-
 	if seq != fsm.seq {
 		recycleMsgBlockBuf(mb.cache.buf)
 		mb.cache.buf = nil
@@ -4067,16 +4040,10 @@ func (mb *msgBlock) readIndexInfo() error {
 
 	// Check if this is a short write index file.
 	if bi < 0 || bi+checksumSize > len(buf) {
-		os.Remove(mb.ifn)
+		defer os.Remove(mb.ifn)
 		return fmt.Errorf("short index file")
 	}
 
-	// Check for consistency if accounting. If something is off bail and we will rebuild.
-	if mb.msgs != (mb.last.seq-mb.first.seq+1)-dmapLen {
-		os.Remove(mb.ifn)
-		return fmt.Errorf("accounting inconsistent")
-	}
-
 	// Checksum
 	copy(mb.lchk[0:], buf[bi:bi+checksumSize])
 	bi += checksumSize
@@ -4184,10 +4151,6 @@ func compareFn(subject string) func(string, string) bool {
 // PurgeEx will remove messages based on subject filters, sequence and number of messages to keep.
 // Will return the number of purged messages.
 func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
-	if sequence > 1 && keep > 0 {
-		return 0, ErrPurgeArgMismatch
-	}
-
 	if subject == _EMPTY_ || subject == fwcs {
 		if keep == 0 && (sequence == 0 || sequence == 1) {
 			return fs.Purge()
@@ -4235,7 +4198,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 			mb.loadMsgsWithLock()
 			shouldExpire = true
 		}
-		if sequence > 1 && sequence <= l {
+		if sequence > 0 && sequence <= l {
 			l = sequence - 1
 		}
 
@@ -4244,13 +4207,10 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 				rl := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 				// Do fast in place remove.
 				// Stats
-				if mb.msgs > 0 {
-					fs.state.Msgs--
-					fs.state.Bytes -= rl
-					mb.msgs--
-					mb.bytes -= rl
-					purged++
-				}
+				fs.state.Msgs--
+				fs.state.Bytes -= rl
+				mb.msgs--
+				mb.bytes -= rl
 				// FSS updates.
 				fs.removePerSubject(sm.subj)
 				mb.removeSeqPerSubject(sm.subj, seq, &smv)
@@ -4271,7 +4231,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 					}
 					mb.dmap[seq] = struct{}{}
 				}
-
+				purged++
 				if maxp > 0 && purged >= maxp {
 					break
 				}
@@ -4431,12 +4391,10 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 			}
 		} else if sm != nil {
 			sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-			if smb.msgs > 0 {
-				smb.bytes -= sz
-				bytes += sz
-				smb.msgs--
-				purged++
-			}
+			smb.bytes -= sz
+			bytes += sz
+			smb.msgs--
+			purged++
 			// Update fss
 			fs.removePerSubject(sm.subj)
 			smb.removeSeqPerSubject(sm.subj, mseq, &smv)
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index a60f61b2..40daa8dd 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -608,7 +608,6 @@ const JSApiConsumerListResponseType = "io.nats.jetstream.api.v1.consumer_list_re
 type JSApiConsumerGetNextRequest struct {
 	Expires   time.Duration `json:"expires,omitempty"`
 	Batch     int           `json:"batch,omitempty"`
-	MaxBytes  int           `json:"max_bytes,omitempty"`
 	NoWait    bool          `json:"no_wait,omitempty"`
 	Heartbeat time.Duration `json:"idle_heartbeat,omitempty"`
 }
@@ -1212,7 +1211,6 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 		}
 		return
 	}
-
 	var cfg StreamConfig
 	if err := json.Unmarshal(msg, &cfg); err != nil {
 		resp.Error = NewJSInvalidJSONError()
@@ -1234,41 +1232,6 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	// If we have a republish directive check if we can create a transform here.
-	if cfg.RePublish != nil {
-		// Check to make sure source is a valid subset of the subjects we have.
-		// Also make sure it does not form a cycle.
-		var srcValid bool
-		for _, subj := range cfg.Subjects {
-			if SubjectsCollide(cfg.RePublish.Source, subj) {
-				srcValid = true
-				break
-			}
-		}
-		if !srcValid {
-			resp.Error = NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish source is not valid subset of subjects"))
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-		var formsCycle bool
-		for _, subj := range cfg.Subjects {
-			if SubjectsCollide(cfg.RePublish.Destination, subj) {
-				formsCycle = true
-				break
-			}
-		}
-		if formsCycle {
-			resp.Error = NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish destination forms a cycle"))
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-		if _, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
-			resp.Error = NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish not valid"))
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-	}
-
 	// Hand off to cluster for processing.
 	if s.JetStreamIsClustered() {
 		s.jsClusteredStreamRequest(ci, acc, subject, reply, rmsg, &cfg)
@@ -3286,13 +3249,6 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subj
 		return
 	}
 
-	// If we are here we are single server mode.
-	if req.Config.Replicas > 1 {
-		resp.Error = NewJSStreamReplicasNotSupportedError()
-		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-		return
-	}
-
 	stream, err := acc.lookupStream(req.Stream)
 	if err != nil {
 		resp.Error = NewJSStreamNotFoundError(Unless(err))
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 689a3b9f..841e2c5c 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1875,11 +1875,10 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			// These are not currently assigned so we will need to do so here.
 			if consumers := mset.getPublicConsumers(); len(consumers) > 0 {
 				for _, o := range consumers {
-					name, cfg := o.String(), o.config()
-					rg := cc.createGroupForConsumer(&cfg, sa)
+					rg := cc.createGroupForConsumer(sa)
 					// Pick a preferred leader.
 					rg.setPreferred()
-
+					name, cfg := o.String(), o.config()
 					// Place our initial state here as well for assignment distribution.
 					state, _ := o.store.State()
 					ca := &consumerAssignment{
@@ -2809,9 +2808,8 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 						js.mu.RUnlock()
 
 						for _, o := range consumers {
+							rg := cc.createGroupForConsumer(sa)
 							name, cfg := o.String(), o.config()
-							rg := cc.createGroupForConsumer(&cfg, sa)
-
 							// Place our initial state here as well for assignment distribution.
 							ca := &consumerAssignment{
 								Group:   rg,
@@ -3141,11 +3139,7 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 
 	if !alreadyRunning {
 		// Process the raft group and make sure its running if needed.
-		storage := mset.config().Storage
-		if ca.Config.MemoryStorage {
-			storage = MemoryStorage
-		}
-		js.createRaftGroup(acc.GetName(), rg, storage)
+		js.createRaftGroup(acc.GetName(), rg, mset.config().Storage)
 	} else {
 		// If we are clustered update the known peers.
 		js.mu.RLock()
@@ -5271,21 +5265,13 @@ func decodeStreamAssignment(buf []byte) (*streamAssignment, error) {
 	return &sa, err
 }
 
-// createGroupForConsumer will create a new group from same peer set as the stream.
-func (cc *jetStreamCluster) createGroupForConsumer(cfg *ConsumerConfig, sa *streamAssignment) *raftGroup {
-	peers := copyStrings(sa.Group.Peers)
+// createGroupForConsumer will create a new group with same peer set as the stream.
+func (cc *jetStreamCluster) createGroupForConsumer(sa *streamAssignment) *raftGroup {
+	peers := sa.Group.Peers
 	if len(peers) == 0 {
 		return nil
 	}
-	if cfg.Replicas > 0 && cfg.Replicas != len(peers) {
-		rand.Shuffle(len(peers), func(i, j int) { peers[i], peers[j] = peers[j], peers[i] })
-		peers = peers[:cfg.Replicas]
-	}
-	storage := sa.Config.Storage
-	if cfg.MemoryStorage {
-		storage = MemoryStorage
-	}
-	return &raftGroup{Name: groupNameForConsumer(peers, storage), Storage: storage, Peers: peers}
+	return &raftGroup{Name: groupNameForConsumer(peers, sa.Config.Storage), Storage: sa.Config.Storage, Peers: peers}
 }
 
 // jsClusteredConsumerRequest is first point of entry to create a consumer with R > 1.
@@ -5386,7 +5372,7 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 
 	// If this is new consumer.
 	if ca == nil {
-		rg := cc.createGroupForConsumer(cfg, sa)
+		rg := cc.createGroupForConsumer(sa)
 		if rg == nil {
 			resp.Error = NewJSInsufficientResourcesError()
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index be507d1a..26114931 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -155,9 +155,6 @@ const (
 	// JSConsumerReplacementWithDifferentNameErr consumer replacement durable config not the same
 	JSConsumerReplacementWithDifferentNameErr ErrorIdentifier = 10106
 
-	// JSConsumerReplicasExceedsStream consumer config replica count exceeds parent stream
-	JSConsumerReplicasExceedsStream ErrorIdentifier = 10126
-
 	// JSConsumerSmallHeartbeatErr consumer idle heartbeat needs to be >= 100ms
 	JSConsumerSmallHeartbeatErr ErrorIdentifier = 10083
 
@@ -433,7 +430,6 @@ var (
 		JSConsumerPullWithRateLimitErr:             {Code: 400, ErrCode: 10086, Description: "consumer in pull mode can not have rate limit set"},
 		JSConsumerPushMaxWaitingErr:                {Code: 400, ErrCode: 10080, Description: "consumer in push mode can not set max waiting"},
 		JSConsumerReplacementWithDifferentNameErr:  {Code: 400, ErrCode: 10106, Description: "consumer replacement durable config not the same"},
-		JSConsumerReplicasExceedsStream:            {Code: 400, ErrCode: 10126, Description: "consumer config replica count exceeds parent stream"},
 		JSConsumerSmallHeartbeatErr:                {Code: 400, ErrCode: 10083, Description: "consumer idle heartbeat needs to be >= 100ms"},
 		JSConsumerStoreFailedErrF:                  {Code: 500, ErrCode: 10104, Description: "error creating store for consumer: {err}"},
 		JSConsumerWQConsumerNotDeliverAllErr:       {Code: 400, ErrCode: 10101, Description: "consumer must be deliver all on workqueue stream"},
@@ -1075,16 +1071,6 @@ func NewJSConsumerReplacementWithDifferentNameError(opts ...ErrorOption) *ApiErr
 	return ApiErrors[JSConsumerReplacementWithDifferentNameErr]
 }
 
-// NewJSConsumerReplicasExceedsStreamError creates a new JSConsumerReplicasExceedsStream error: "consumer config replica count exceeds parent stream"
-func NewJSConsumerReplicasExceedsStreamError(opts ...ErrorOption) *ApiError {
-	eopts := parseOpts(opts)
-	if ae, ok := eopts.err.(*ApiError); ok {
-		return ae
-	}
-
-	return ApiErrors[JSConsumerReplicasExceedsStream]
-}
-
 // NewJSConsumerSmallHeartbeatError creates a new JSConsumerSmallHeartbeatErr error: "consumer idle heartbeat needs to be >= 100ms"
 func NewJSConsumerSmallHeartbeatError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/memstore.go b/server/memstore.go
index 65493301..d06cad4f 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -428,10 +428,6 @@ func (ms *memStore) expireMsgs() {
 // PurgeEx will remove messages based on subject filters, sequence and number of messages to keep.
 // Will return the number of purged messages.
 func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
-	if sequence > 1 && keep > 0 {
-		return 0, ErrPurgeArgMismatch
-	}
-
 	if subject == _EMPTY_ || subject == fwcs {
 		if keep == 0 && (sequence == 0 || sequence == 1) {
 			return ms.Purge()
@@ -459,7 +455,7 @@ func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint6
 			ss.Msgs -= keep
 		}
 		last := ss.Last
-		if sequence > 1 {
+		if sequence > 0 {
 			last = sequence - 1
 		}
 		ms.mu.Lock()
diff --git a/server/raft.go b/server/raft.go
index e78c15cf..39163a43 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -228,11 +228,11 @@ type lps struct {
 }
 
 const (
-	minElectionTimeoutDefault      = 2 * time.Second
-	maxElectionTimeoutDefault      = 5 * time.Second
+	minElectionTimeoutDefault      = 4 * time.Second
+	maxElectionTimeoutDefault      = 9 * time.Second
 	minCampaignTimeoutDefault      = 100 * time.Millisecond
 	maxCampaignTimeoutDefault      = 8 * minCampaignTimeoutDefault
-	hbIntervalDefault              = 500 * time.Millisecond
+	hbIntervalDefault              = 1 * time.Second
 	lostQuorumIntervalDefault      = hbIntervalDefault * 20 // 10 seconds
 	lostQuorumCheckIntervalDefault = hbIntervalDefault * 20 // 10 seconds
 
@@ -2606,12 +2606,6 @@ func (n *raft) handleAppendEntry(sub *subscription, c *client, _ *Account, subje
 	} else {
 		n.warn("AppendEntry failed to be placed on internal channel: corrupt entry")
 	}
-	n.Lock()
-	// Don't reset here if we have been asked to assume leader position.
-	if !n.lxfer {
-		n.resetElectionTimeout()
-	}
-	n.Unlock()
 }
 
 // Lock should be held.
@@ -2680,6 +2674,11 @@ func (n *raft) updateLeader(newLeader string) {
 func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	n.Lock()
 
+	// Don't reset here if we have been asked to assume leader position.
+	if !n.lxfer {
+		n.resetElectionTimeout()
+	}
+
 	// Just return if closed or we had previous write error.
 	if n.state == Closed || n.werr != nil {
 		n.Unlock()
diff --git a/server/store.go b/server/store.go
index e6158a0c..10fc10a7 100644
--- a/server/store.go
+++ b/server/store.go
@@ -61,8 +61,6 @@ var (
 	ErrInvalidSequence = errors.New("invalid sequence")
 	// ErrSequenceMismatch is returned when storing a raw message and the expected sequence is wrong.
 	ErrSequenceMismatch = errors.New("expected sequence does not match store")
-	// ErrPurgeArgMismatch is returned when PurgeEx is called with sequence > 1 and keep > 0.
-	ErrPurgeArgMismatch = errors.New("sequence > 1 && keep > 0 not allowed")
 )
 
 // StoreMsg is the stored message format for messages that are retained by the Store layer.
diff --git a/server/stream.go b/server/stream.go
index 7efdb58b..0b238331 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -58,9 +58,6 @@ type StreamConfig struct {
 	Mirror       *StreamSource   `json:"mirror,omitempty"`
 	Sources      []*StreamSource `json:"sources,omitempty"`
 
-	// Allow republish of the message after being sequenced and stored.
-	RePublish *SubjectMapping `json:"republish,omitempty"`
-
 	// Optional qualifiers. These can not be modified after set to true.
 
 	// Sealed will seal a stream so no messages can get out or in.
@@ -74,12 +71,6 @@ type StreamConfig struct {
 	AllowRollup bool `json:"allow_rollup_hdrs"`
 }
 
-// SubjectMapping allows a source subject to be mapped to a destination subject for republishing.
-type SubjectMapping struct {
-	Source      string `json:"src,omitempty"`
-	Destination string `json:"dest"`
-}
-
 // JSPubAckResponse is a formal response to a publish operation.
 type JSPubAckResponse struct {
 	Error *ApiError `json:"error,omitempty"`
@@ -211,9 +202,6 @@ type stream struct {
 	// Indicates we have direct consumers.
 	directs int
 
-	// For republishing.
-	tr *transform
-
 	// TODO(dlc) - Hide everything below behind two pointers.
 	// Clustered mode.
 	sa       *streamAssignment
@@ -264,13 +252,6 @@ const (
 	JSResponseType        = "Nats-Response-Type"
 )
 
-// Headers for republished messages.
-const (
-	JSStream       = "Nats-Stream"
-	JSSequence     = "Nats-Sequence"
-	JSLastSequence = "Nats-Last-Sequence"
-)
-
 // Rollups, can be subject only or all messages.
 const (
 	JSMsgRollupSubject = "sub"
@@ -415,17 +396,6 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		mset.ackq = s.newIPQueue(qpfx + "acks") // of uint64
 	}
 
-	// Check for RePublish.
-	if cfg.RePublish != nil {
-		tr, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
-		if err != nil {
-			jsa.mu.Unlock()
-			return nil, fmt.Errorf("stream configuration for republish not valid")
-		}
-		// Assign our transform for republishing.
-		mset.tr = tr
-	}
-
 	jsa.streams[cfg.Name] = mset
 	storeDir := filepath.Join(jsa.storeDir, streamsDir, cfg.Name)
 	jsa.mu.Unlock()
@@ -950,8 +920,7 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicates window can not be larger then max age"))
 	}
 	if lim.Duplicates > 0 && cfg.Duplicates > lim.Duplicates {
-		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicates window can not be larger then server limit of %v",
-			lim.Duplicates.String()))
+		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicates window can not be larger then server limit of %v", lim.Duplicates.String()))
 	}
 	if cfg.Duplicates > 0 && cfg.Duplicates < 100*time.Millisecond {
 		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicates window needs to be >= 100ms"))
@@ -1139,35 +1108,6 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		return StreamConfig{}, NewJSStreamMaxStreamBytesExceededError()
 	}
 
-	// If we have a republish directive check if we can create a transform here.
-	if cfg.RePublish != nil {
-		// Check to make sure source is a valid subset of the subjects we have.
-		// Also make sure it does not form a cycle.
-		var srcValid bool
-		for _, subj := range cfg.Subjects {
-			if SubjectsCollide(cfg.RePublish.Source, subj) {
-				srcValid = true
-				break
-			}
-		}
-		if !srcValid {
-			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish source is not valid subset of subjects"))
-		}
-		var formsCycle bool
-		for _, subj := range cfg.Subjects {
-			if SubjectsCollide(cfg.RePublish.Destination, subj) {
-				formsCycle = true
-				break
-			}
-		}
-		if formsCycle {
-			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish destination forms a cycle"))
-		}
-		if _, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
-			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish not valid"))
-		}
-	}
-
 	return cfg, nil
 }
 
@@ -2305,6 +2245,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 			req.Config.OptStartTime = &si.start
 			req.Config.DeliverPolicy = DeliverByStartTime
 		}
+
 	} else {
 		req.Config.OptStartSeq = seq
 		req.Config.DeliverPolicy = DeliverByStartSequence
@@ -2731,18 +2672,14 @@ func (mset *stream) startingSequenceForSources() {
 	var state StreamState
 	mset.store.FastState(&state)
 
-	// If the last time has been stamped remember in case we need to fall back to this for any given upstream source.
+	// If we have no messages fall back to start by time for now.
 	// TODO(dlc) - This will be ok, but should formalize with new approach and more formal and durable state.
-	if !state.LastTime.IsZero() {
+	if state.Msgs == 0 {
 		for _, si := range mset.sources {
 			si.start = state.LastTime
 		}
-	}
-	// Bail if no messages, meaning no context.
-	if state.Msgs == 0 {
 		return
 	}
-
 	// For short circuiting return.
 	expected := len(mset.cfg.Sources)
 	seqs := make(map[string]uint64)
@@ -3466,28 +3403,11 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	clfs := mset.clfs
 	mset.lseq++
 	tierName := mset.tier
-
-	// Republish state if needed.
-	var tsubj string
-	var tlseq uint64
-	if mset.tr != nil {
-		tsubj, _ = mset.tr.transformSubject(subject)
-	}
-	republish := tsubj != _EMPTY_
-
 	// We hold the lock to this point to make sure nothing gets between us since we check for pre-conditions.
 	// Currently can not hold while calling store b/c we have inline storage update calls that may need the lock.
 	// Note that upstream that sets seq/ts should be serialized as much as possible.
 	mset.mu.Unlock()
 
-	// If we are republishing grab last sequence for this exact subject. Aids in gap detection for lightweight clients.
-	if republish {
-		var smv StoreMsg
-		if sm, _ := store.LoadLastMsg(subject, &smv); sm != nil {
-			tlseq = sm.seq
-		}
-	}
-
 	// Store actual msg.
 	if lseq == 0 && ts == 0 {
 		seq, ts, err = store.StoreMsg(subject, hdr, msg)
@@ -3555,13 +3475,6 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 			response = append(pubAck, strconv.FormatUint(seq, 10)...)
 			response = append(response, '}')
 		}
-		// Check for republish.
-		if republish {
-			hdr = genHeader(hdr, JSStream, name)
-			hdr = genHeader(hdr, JSSequence, strconv.FormatUint(seq, 10))
-			hdr = genHeader(hdr, JSLastSequence, strconv.FormatUint(tlseq, 10))
-			mset.outq.send(newJSPubMsg(tsubj, subject, _EMPTY_, copyBytes(hdr), copyBytes(msg), nil, seq))
-		}
 	}
 
 	// Send response here.
diff --git a/server/util.go b/server/util.go
index c7f5827e..24cef1ac 100644
--- a/server/util.go
+++ b/server/util.go
@@ -314,9 +314,9 @@ func getURLsAsString(urls []*url.URL) []string {
 }
 
 // copyBytes make a new slice of the same size than `src` and copy its content.
-// If `src` is nil or its length is 0, then this returns `nil`
+// If `src` is nil, then this returns `nil`
 func copyBytes(src []byte) []byte {
-	if len(src) == 0 {
+	if src == nil {
 		return nil
 	}
 	dst := make([]byte, len(src))


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 9249199c..d03eaa68 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3825,85 +3825,3 @@ func TestFileStoreRememberLastMsgTime(t *testing.T) {
 	require_True(t, lt == fs.State().LastTime)
 	require_True(t, seq == 6)
 }
-
-func (fs *fileStore) getFirstBlock() *msgBlock {
-	fs.mu.RLock()
-	defer fs.mu.RUnlock()
-	if len(fs.blks) == 0 {
-		return nil
-	}
-	return fs.blks[0]
-}
-
-func TestFileStoreRebuildStateDmapAccountingBug(t *testing.T) {
-	storeDir := createDir(t, JetStreamStoreDir)
-	defer removeDir(t, storeDir)
-
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir, BlockSize: 1024 * 1024},
-		StreamConfig{Name: "TEST", Storage: FileStorage},
-	)
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	for i := 0; i < 100; i++ {
-		_, _, err = fs.StoreMsg("foo", nil, nil)
-		require_NoError(t, err)
-	}
-	// Delete 2-40.
-	for i := 2; i <= 40; i++ {
-		_, err := fs.RemoveMsg(uint64(i))
-		require_NoError(t, err)
-	}
-
-	mb := fs.getFirstBlock()
-	require_True(t, mb != nil)
-
-	check := func() {
-		t.Helper()
-		mb.mu.RLock()
-		defer mb.mu.RUnlock()
-		dmapLen := uint64(len(mb.dmap))
-		if mb.msgs != (mb.last.seq-mb.first.seq+1)-dmapLen {
-			t.Fatalf("Consistency check failed: %d != %d -> last %d first %d len(dmap) %d",
-				mb.msgs, (mb.last.seq-mb.first.seq+1)-dmapLen, mb.last.seq, mb.first.seq, dmapLen)
-		}
-	}
-
-	check()
-
-	mb.mu.Lock()
-	mb.compact()
-	mb.mu.Unlock()
-
-	// Now delete first.
-	_, err = fs.RemoveMsg(1)
-	require_NoError(t, err)
-
-	mb.mu.Lock()
-	_, err = mb.rebuildStateLocked()
-	require_NoError(t, err)
-	mb.mu.Unlock()
-
-	check()
-}
-
-func TestFileStorePurgeExWithSubject(t *testing.T) {
-	storeDir := createDir(t, JetStreamStoreDir)
-	defer removeDir(t, storeDir)
-
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "TEST", Storage: FileStorage},
-	)
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	for i := 0; i < 100; i++ {
-		_, _, err = fs.StoreMsg("foo", nil, nil)
-		require_NoError(t, err)
-	}
-
-	// This should purge all.
-	fs.PurgeEx("foo", 1, 0)
-	require_True(t, fs.State().Msgs == 0)
-}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 4de1e8f9..826100c9 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -3996,6 +3996,13 @@ func TestJetStreamClusterNoDuplicateOnNodeRestart(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
+	sl := c.streamLeader("$G", "TEST")
+	if s == sl {
+		nc.Close()
+		nc, js = jsClientConnect(t, s)
+		defer nc.Close()
+	}
+
 	sub, err := js.SubscribeSync("foo", nats.Durable("dlc"))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -4008,14 +4015,14 @@ func TestJetStreamClusterNoDuplicateOnNodeRestart(t *testing.T) {
 		m.AckSync()
 	}
 
-	sl := c.streamLeader("$G", "TEST")
 	sl.Shutdown()
 	c.restartServer(sl)
 	c.waitOnStreamLeader("$G", "TEST")
+	c.waitOnConsumerLeader("$G", "TEST", "dlc")
 
 	// Send second msg
 	js.Publish("foo", []byte("msg2"))
-	msg, err := sub.NextMsg(time.Second)
+	msg, err := sub.NextMsg(5 * time.Second)
 	if err != nil {
 		t.Fatalf("Error getting message: %v", err)
 	}
@@ -10528,153 +10535,3 @@ func TestJetStreamClusterNewHealthz(t *testing.T) {
 	sl = c.restartServer(sl)
 	c.waitOnServerHealthz(sl)
 }
-
-func TestJetStreamClusterConsumerOverrides(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "JSC", 3)
-	defer c.shutdown()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{"foo"},
-		Replicas: 3,
-	})
-	require_NoError(t, err)
-
-	// Test replica override.
-	// Make sure we can not go "wider" than the parent stream.
-	ccReq := CreateConsumerRequest{
-		Stream: "TEST",
-		Config: ConsumerConfig{
-			Durable:   "d",
-			AckPolicy: AckExplicit,
-			Replicas:  5,
-		},
-	}
-	req, err := json.Marshal(ccReq)
-	require_NoError(t, err)
-
-	ci, err := nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", "d"), req, time.Second)
-	require_NoError(t, err)
-
-	var resp JSApiConsumerCreateResponse
-	err = json.Unmarshal(ci.Data, &resp)
-	require_NoError(t, err)
-
-	if resp.Error == nil || !IsNatsErr(resp.Error, JSConsumerReplicasExceedsStream) {
-		t.Fatalf("Expected an error when replicas > parent stream, got %+v", resp.Error)
-	}
-
-	// Durables inherit the replica count from the stream, so make sure we can override that.
-	ccReq.Config.Replicas = 1
-	req, err = json.Marshal(ccReq)
-	require_NoError(t, err)
-
-	ci, err = nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", "d"), req, time.Second)
-	require_NoError(t, err)
-
-	resp.Error = nil
-	err = json.Unmarshal(ci.Data, &resp)
-	require_NoError(t, err)
-	require_True(t, resp.Error == nil)
-
-	checkCount := func(durable string, expected int) {
-		t.Helper()
-		count := 0
-		for _, s := range c.servers {
-			if mset, err := s.GlobalAccount().lookupStream("TEST"); err == nil {
-				if o := mset.lookupConsumer(durable); o != nil {
-					count++
-				}
-			}
-		}
-		if count != expected {
-			t.Fatalf("Expected %d consumers in cluster, got %d", expected, count)
-		}
-	}
-	checkCount("d", 1)
-
-	// Now override storage and force storage to memory based.
-	ccReq.Config.MemoryStorage = true
-	ccReq.Config.Durable = "m"
-	ccReq.Config.Replicas = 2
-
-	req, err = json.Marshal(ccReq)
-	require_NoError(t, err)
-
-	ci, err = nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", "m"), req, time.Second)
-	require_NoError(t, err)
-
-	resp.Error = nil
-	err = json.Unmarshal(ci.Data, &resp)
-	require_NoError(t, err)
-	require_True(t, resp.Error == nil)
-
-	checkCount("m", 2)
-}
-
-func TestJetStreamClusterStreamRepublish(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "JSC", 3)
-	defer c.shutdown()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	// Do by hand for now.
-	cfg := &StreamConfig{
-		Name:     "RP",
-		Storage:  MemoryStorage,
-		Subjects: []string{"foo", "bar", "baz"},
-		Replicas: 1,
-		RePublish: &SubjectMapping{
-			Source:      ">",
-			Destination: "RP.>",
-		},
-	}
-
-	addStream(t, nc, cfg)
-
-	sub, err := nc.SubscribeSync("RP.>")
-	require_NoError(t, err)
-
-	msg, toSend := []byte("OK TO REPUBLISH?"), 100
-	for i := 0; i < toSend; i++ {
-		_, err = js.PublishAsync("foo", msg)
-		require_NoError(t, err)
-		_, err = js.PublishAsync("bar", msg)
-		require_NoError(t, err)
-		_, err = js.PublishAsync("baz", msg)
-		require_NoError(t, err)
-	}
-	select {
-	case <-js.PublishAsyncComplete():
-	case <-time.After(5 * time.Second):
-		t.Fatalf("Did not receive completion signal")
-	}
-
-	checkSubsPending(t, sub, toSend*3)
-
-	lseq := map[string]int{
-		"foo": 0,
-		"bar": 0,
-		"baz": 0,
-	}
-
-	for i := 1; i <= toSend; i++ {
-		m, err := sub.NextMsg(time.Second)
-		require_NoError(t, err)
-		// Grab info from Header
-		require_True(t, m.Header.Get(JSStream) == "RP")
-		// Make sure sequence is correct.
-		seq, err := strconv.Atoi(m.Header.Get(JSSequence))
-		require_NoError(t, err)
-		require_True(t, seq == i)
-		// Make sure last sequence matches last seq we received on this subject.
-		last, err := strconv.Atoi(m.Header.Get(JSLastSequence))
-		require_NoError(t, err)
-		require_True(t, last == lseq[m.Subject])
-		lseq[m.Subject] = seq
-	}
-}
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index 21129d5e..ef168b99 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -424,7 +424,7 @@ func (sc *supercluster) leader() *Server {
 
 func (sc *supercluster) waitOnLeader() {
 	sc.t.Helper()
-	expires := time.Now().Add(5 * time.Second)
+	expires := time.Now().Add(10 * time.Second)
 	for time.Now().Before(expires) {
 		for _, c := range sc.clusters {
 			if leader := c.leader(); leader != nil {
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 7a3c8023..ac66fd91 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -650,7 +650,7 @@ func TestJetStreamConsumerMaxDeliveries(t *testing.T) {
 
 func TestJetStreamNextReqFromMsg(t *testing.T) {
 	bef := time.Now()
-	expires, _, _, _, _, _, err := nextReqFromMsg([]byte(`{"expires":5000000000}`)) // nanoseconds
+	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)) {
@@ -14435,16 +14435,15 @@ func TestJetStreamPullConsumerRequestMaximums(t *testing.T) {
 	require_NoError(t, err)
 
 	_, err = mset.addConsumer(&ConsumerConfig{
-		Durable:            "dlc",
-		MaxRequestBatch:    10,
-		MaxRequestMaxBytes: 10_000,
-		MaxRequestExpires:  time.Second,
-		AckPolicy:          AckExplicit,
+		Durable:           "dlc",
+		MaxRequestBatch:   10,
+		MaxRequestExpires: time.Second,
+		AckPolicy:         AckExplicit,
 	})
 	require_NoError(t, err)
 
-	genReq := func(b, mb int, e time.Duration) []byte {
-		req := &JSApiConsumerGetNextRequest{Batch: b, Expires: e, MaxBytes: mb}
+	genReq := func(b int, e time.Duration) []byte {
+		req := &JSApiConsumerGetNextRequest{Batch: b, Expires: e}
 		jreq, err := json.Marshal(req)
 		require_NoError(t, err)
 		return jreq
@@ -14453,21 +14452,14 @@ func TestJetStreamPullConsumerRequestMaximums(t *testing.T) {
 	rsubj := fmt.Sprintf(JSApiRequestNextT, "TEST", "dlc")
 
 	// Exceeds max batch size.
-	resp, err := nc.Request(rsubj, genReq(11, 0, 100*time.Millisecond), time.Second)
+	resp, err := nc.Request(rsubj, genReq(11, 100*time.Millisecond), time.Second)
 	require_NoError(t, err)
 	if status := resp.Header.Get("Status"); status != "409" {
 		t.Fatalf("Expected a 409 status code, got %q", status)
 	}
 
 	// Exceeds max expires.
-	resp, err = nc.Request(rsubj, genReq(1, 0, 10*time.Minute), time.Second)
-	require_NoError(t, err)
-	if status := resp.Header.Get("Status"); status != "409" {
-		t.Fatalf("Expected a 409 status code, got %q", status)
-	}
-
-	// Exceeds max bytes.
-	resp, err = nc.Request(rsubj, genReq(10, 10_000*2, 10*time.Minute), time.Second)
+	resp, err = nc.Request(rsubj, genReq(1, 10*time.Minute), time.Second)
 	require_NoError(t, err)
 	if status := resp.Header.Get("Status"); status != "409" {
 		t.Fatalf("Expected a 409 status code, got %q", status)
@@ -17273,273 +17265,6 @@ func TestJetStreamWorkQueueSourceRestart(t *testing.T) {
 	}
 }
 
-func TestJetStreamWorkQueueSourceNamingRestart(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	if config := s.JetStreamConfig(); config != nil {
-		defer removeDir(t, config.StoreDir)
-	}
-	defer s.Shutdown()
-
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{Name: "C1", Subjects: []string{"foo.*"}})
-	require_NoError(t, err)
-	_, err = js.AddStream(&nats.StreamConfig{Name: "C2", Subjects: []string{"bar.*"}})
-	require_NoError(t, err)
-
-	sendCount := 10
-	for i := 0; i < sendCount; i++ {
-		_, err = js.Publish(fmt.Sprintf("foo.%d", i), nil)
-		require_NoError(t, err)
-		_, err = js.Publish(fmt.Sprintf("bar.%d", i), nil)
-		require_NoError(t, err)
-	}
-
-	// TODO Test will always pass if pending is 0
-	pending := 1
-	// For some yet unknown reason this failure seems to require 2 streams to source from.
-	// This might possibly be timing, as the test sometimes passes
-	streams := 2
-	totalPending := uint64(streams * pending)
-	totalMsgs := streams * sendCount
-	totalNonPending := streams * (sendCount - pending)
-
-	// TODO Test will always pass if this is named A (go returns directory names sorted)
-	// A: this stream is recovered BEFORE C1/C2, tbh, I'd expect this to be the case to fail, but it isn't
-	// D: this stream is recovered AFTER C1/C2, which is the case that fails (perhaps it is timing)
-	srcName := "D"
-	_, err = js.AddStream(&nats.StreamConfig{
-		Name:      srcName,
-		Retention: nats.WorkQueuePolicy,
-		Sources:   []*nats.StreamSource{{Name: "C1"}, {Name: "C2"}},
-	})
-	require_NoError(t, err)
-
-	// Add a consumer and consume all but totalPending messages
-	_, err = js.AddConsumer(srcName, &nats.ConsumerConfig{Durable: "dur", AckPolicy: nats.AckExplicitPolicy})
-	require_NoError(t, err)
-
-	sub, err := js.PullSubscribe("", "dur", nats.BindStream(srcName))
-	require_NoError(t, err)
-
-	checkFor(t, 5*time.Second, time.Millisecond*200, func() error {
-		if ci, err := js.ConsumerInfo(srcName, "dur"); err != nil {
-			return err
-		} else if ci.NumPending != uint64(totalMsgs) {
-			return fmt.Errorf("not enough messages: %d", ci.NumPending)
-		}
-		return nil
-	})
-
-	// consume all but messages we want pending
-	msgs, err := sub.Fetch(totalNonPending)
-	require_NoError(t, err)
-	require_True(t, len(msgs) == totalNonPending)
-
-	for _, m := range msgs {
-		err = m.AckSync()
-		require_NoError(t, err)
-	}
-
-	ci, err := js.ConsumerInfo(srcName, "dur")
-	require_NoError(t, err)
-	require_True(t, ci.NumPending == totalPending)
-
-	si, err := js.StreamInfo(srcName)
-	require_NoError(t, err)
-	require_True(t, si.State.Msgs == totalPending)
-
-	// Restart server
-	nc.Close()
-	sd := s.JetStreamConfig().StoreDir
-	s.Shutdown()
-	time.Sleep(200 * time.Millisecond)
-	s = RunJetStreamServerOnPort(-1, sd)
-	defer s.Shutdown()
-
-	checkFor(t, 10*time.Second, 200*time.Millisecond, func() error {
-		hs := s.healthz()
-		if hs.Status == "ok" && hs.Error == _EMPTY_ {
-			return nil
-		}
-		return fmt.Errorf("healthz %s %s", hs.Error, hs.Status)
-	})
-
-	nc, js = jsClientConnect(t, s)
-	defer nc.Close()
-
-	si, err = js.StreamInfo(srcName)
-	require_NoError(t, err)
-
-	if si.State.Msgs != totalPending {
-		t.Fatalf("Expected 0 messages on restart, got %d", si.State.Msgs)
-	}
-}
-
-func TestJetStreamPullMaxBytes(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	if config := s.JetStreamConfig(); config != nil {
-		defer removeDir(t, config.StoreDir)
-	}
-	defer s.Shutdown()
-
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name: "TEST",
-	})
-	require_NoError(t, err)
-
-	// Put in ~2MB, each ~100k
-	msz := 99_980
-	total, msg := 20, []byte(strings.Repeat("Z", msz))
-
-	for i := 0; i < total; i++ {
-		if _, err := js.Publish("TEST", msg); err != nil {
-			t.Fatalf("Unexpected publish error: %v", err)
-		}
-	}
-
-	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
-		Durable:   "pr",
-		AckPolicy: nats.AckExplicitPolicy,
-	})
-	require_NoError(t, err)
-
-	req := &JSApiConsumerGetNextRequest{MaxBytes: 100, NoWait: true}
-	jreq, _ := json.Marshal(req)
-
-	subj := fmt.Sprintf(JSApiRequestNextT, "TEST", "pr")
-	reply := "_pr_"
-	sub, _ := nc.SubscribeSync(reply)
-	defer sub.Unsubscribe()
-
-	checkHeader := func(m *nats.Msg, expected *nats.Header) {
-		t.Helper()
-		if len(m.Data) != 0 {
-			t.Fatalf("Did not expect data, got %d bytes", len(m.Data))
-		}
-		if !reflect.DeepEqual(&m.Header, expected) {
-			t.Fatalf("Expected %+v hdr, got %+v", expected, m.Header)
-		}
-	}
-
-	// If we ask for less MaxBytes then a single message make sure we get an error.
-	badReq := &nats.Header{"Status": []string{"408"}, "Description": []string{"Message Size Exceeds MaxBytes"}}
-
-	nc.PublishRequest(subj, reply, jreq)
-	m, err := sub.NextMsg(time.Second)
-	require_NoError(t, err)
-	checkSubsPending(t, sub, 0)
-	checkHeader(m, badReq)
-
-	// If we request a ton of max bytes make sure batch size overrides.
-	req = &JSApiConsumerGetNextRequest{Batch: 1, MaxBytes: 10_000_000, NoWait: true}
-	jreq, _ = json.Marshal(req)
-	nc.PublishRequest(subj, reply, jreq)
-	checkSubsPending(t, sub, 1)
-
-	m, err = sub.NextMsg(time.Second)
-	require_NoError(t, err)
-	require_True(t, len(m.Data) == msz)
-	require_True(t, len(m.Header) == 0)
-	checkSubsPending(t, sub, 0)
-
-	// Same but with batch > 1
-	req = &JSApiConsumerGetNextRequest{Batch: 5, MaxBytes: 10_000_000, NoWait: true}
-	jreq, _ = json.Marshal(req)
-	nc.PublishRequest(subj, reply, jreq)
-	checkSubsPending(t, sub, 5)
-	for i := 0; i < 5; i++ {
-		m, err = sub.NextMsg(time.Second)
-		require_NoError(t, err)
-		require_True(t, len(m.Data) == msz)
-		require_True(t, len(m.Header) == 0)
-	}
-	checkSubsPending(t, sub, 0)
-
-	// Now ask for large batch but make sure we are limited by batch size.
-	req = &JSApiConsumerGetNextRequest{Batch: 1_000, MaxBytes: msz * 5, NoWait: true}
-	jreq, _ = json.Marshal(req)
-	nc.PublishRequest(subj, reply, jreq)
-	checkSubsPending(t, sub, 5)
-	for i := 0; i < 5; i++ {
-		m, err = sub.NextMsg(time.Second)
-		require_NoError(t, err)
-		require_True(t, len(m.Data) == msz)
-		require_True(t, len(m.Header) == 0)
-	}
-	checkSubsPending(t, sub, 0)
-
-	req = &JSApiConsumerGetNextRequest{Batch: 1_000, MaxBytes: msz + 20, NoWait: true}
-	jreq, _ = json.Marshal(req)
-	nc.PublishRequest(subj, reply, jreq)
-	checkSubsPending(t, sub, 1)
-	m, err = sub.NextMsg(time.Second)
-	require_NoError(t, err)
-	require_True(t, len(m.Data) == msz)
-	require_True(t, len(m.Header) == 0)
-	checkSubsPending(t, sub, 0)
-}
-
-func TestJetStreamStreamRepublishCycle(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	if config := s.JetStreamConfig(); config != nil {
-		defer removeDir(t, config.StoreDir)
-	}
-	defer s.Shutdown()
-
-	nc, _ := jsClientConnect(t, s)
-	defer nc.Close()
-
-	// Do by hand for now.
-	cfg := &StreamConfig{
-		Name:     "RPC",
-		Storage:  MemoryStorage,
-		Subjects: []string{"foo.>", "bar.*", "baz"},
-	}
-
-	expectFail := func() {
-		t.Helper()
-		req, err := json.Marshal(cfg)
-		require_NoError(t, err)
-		rmsg, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
-		require_NoError(t, err)
-		var resp JSApiStreamCreateResponse
-		err = json.Unmarshal(rmsg.Data, &resp)
-		require_NoError(t, err)
-		if resp.Type != JSApiStreamCreateResponseType {
-			t.Fatalf("Invalid response type %s expected %s", resp.Type, JSApiStreamCreateResponseType)
-		}
-		if resp.Error == nil {
-			t.Fatalf("Expected error but got none")
-		}
-		if !strings.Contains(resp.Error.Description, "republish destination forms a cycle") {
-			t.Fatalf("Expected cycle error, got %q", resp.Error.Description)
-		}
-	}
-
-	cfg.RePublish = &SubjectMapping{
-		Source:      "foo.>",
-		Destination: "foo.>",
-	}
-	expectFail()
-
-	cfg.RePublish = &SubjectMapping{
-		Source:      "bar.bar",
-		Destination: "foo.bar",
-	}
-	expectFail()
-
-	cfg.RePublish = &SubjectMapping{
-		Source:      "baz",
-		Destination: "bar.bar",
-	}
-	expectFail()
-}
-
 ///////////////////////////////////////////////////////////////////////////
 // Simple JetStream Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 92fa00c5..413d39a7 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -391,17 +391,3 @@ func TestMemStoreStreamTruncate(t *testing.T) {
 		t.Fatalf("Expected deleted to be %+v, got %+v\n", expected, state.Deleted)
 	}
 }
-
-func TestMemStorePurgeExWithSubject(t *testing.T) {
-	ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage})
-	require_NoError(t, err)
-
-	for i := 0; i < 100; i++ {
-		_, _, err = ms.StoreMsg("foo", nil, nil)
-		require_NoError(t, err)
-	}
-
-	// This should purge all.
-	ms.PurgeEx("foo", 1, 0)
-	require_True(t, ms.State().Msgs == 0)
-}