Repository: nats-io/nats-server
Pull request #2387 (2021-07-30 15:49:22)

[IMPROVED] Server restart time with many expired JetStream messages.
When a server was restarting, if a stream had a MaxAge configured and there were a very large amount of messages to expire, this process would take too long and delay the server reaching a running state.

This PR introduces an optimized expiration tailored for startup.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index d63105bd..4ce0004d 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -86,10 +86,8 @@ const (
 	DeliverNew
 	// DeliverByStartSequence will look for a defined starting sequence to start.
 	DeliverByStartSequence
-	// DeliverByStartTime will select the first messsage with a timestamp >= to StartTime.
+	// DeliverByStartTime will select the first messsage with a timestamp >= to StartTime
 	DeliverByStartTime
-	// DeliverLastPerSubject will start the consumer with the last message for all subjects received.
-	DeliverLastPerSubject
 )
 
 func (dp DeliverPolicy) String() string {
@@ -104,8 +102,6 @@ func (dp DeliverPolicy) String() string {
 		return "by_start_sequence"
 	case DeliverByStartTime:
 		return "by_start_time"
-	case DeliverLastPerSubject:
-		return "last_per_subject"
 	default:
 		return "undefined"
 	}
@@ -190,7 +186,6 @@ type consumer struct {
 	asflr             uint64
 	sgap              uint64
 	dsubj             string
-	lss               *lastSeqSkipList
 	rlimit            *rate.Limiter
 	reqSub            *subscription
 	ackSub            *subscription
@@ -367,68 +362,38 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	switch config.DeliverPolicy {
 	case DeliverAll:
 		if config.OptStartSeq > 0 {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver all, but optional start sequence is also set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver all, but optional start sequence is also set")
 		}
 		if config.OptStartTime != nil {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver all, but optional start time is also set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver all, but optional start time is also set")
 		}
 	case DeliverLast:
 		if config.OptStartSeq > 0 {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver last, but optional start sequence is also set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver last, but optional start sequence is also set")
 		}
 		if config.OptStartTime != nil {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver last, but optional start time is also set")
-		}
-	case DeliverLastPerSubject:
-		if config.OptStartSeq > 0 {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver last per subject, but optional start sequence is also set")
-		}
-		if config.OptStartTime != nil {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver last per subject, but optional start time is also set")
-		}
-		badConfig := config.FilterSubject == _EMPTY_
-		if !badConfig {
-			subjects, ext := mset.allSubjects()
-			if len(subjects) == 1 && !ext && subjects[0] == config.FilterSubject && subjectIsLiteral(subjects[0]) {
-				badConfig = true
-			}
-		}
-		if badConfig {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver last per subject, but filter subject is not set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver last, but optional start time is also set")
 		}
 	case DeliverNew:
 		if config.OptStartSeq > 0 {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver new, but optional start sequence is also set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver new, but optional start sequence is also set")
 		}
 		if config.OptStartTime != nil {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver new, but optional start time is also set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver new, but optional start time is also set")
 		}
 	case DeliverByStartSequence:
 		if config.OptStartSeq == 0 {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver by start sequence, but optional start sequence is not set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver by start sequence, but optional start sequence is not set")
 		}
 		if config.OptStartTime != nil {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver by start sequence, but optional start time is also set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver by start sequence, but optional start time is also set")
 		}
 	case DeliverByStartTime:
 		if config.OptStartTime == nil {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver by start time, but optional start time is not set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver by start time, but optional start time is not set")
 		}
 		if config.OptStartSeq != 0 {
-			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver by start time, but optional start sequence is also set")
+			return nil, ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}", "consumer delivery policy is deliver by start time, but optional start sequence is also set")
 		}
 	}
 
@@ -1965,16 +1930,7 @@ func (o *consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dc ui
 	}
 	for {
 		seq, dc := o.sseq, uint64(1)
-		if o.hasSkipListPending() {
-			seq = o.lss.seqs[0]
-			if len(o.lss.seqs) == 1 {
-				o.sseq = o.lss.resume
-				o.lss = nil
-				o.updateSkipped()
-			} else {
-				o.lss.seqs = o.lss.seqs[1:]
-			}
-		} else if o.hasRedeliveries() {
+		if o.hasRedeliveries() {
 			seq = o.getNextToRedeliver()
 			dc = o.incDeliveryCount(seq)
 			if o.maxdc > 0 && dc > o.maxdc {
@@ -2612,34 +2568,12 @@ func ackReplyInfo(subject string) (sseq, dseq, dc uint64) {
 
 // NextSeq returns the next delivered sequence number for this consumer.
 func (o *consumer) nextSeq() uint64 {
-	o.mu.RLock()
+	o.mu.Lock()
 	dseq := o.dseq
-	o.mu.RUnlock()
+	o.mu.Unlock()
 	return dseq
 }
 
-// Used to hold skip list when deliver policy is last per subject.
-type lastSeqSkipList struct {
-	resume uint64
-	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 {
-	return o.lss != nil && len(o.lss.seqs) > 0
-}
-
 // Will select the starting sequence.
 func (o *consumer) selectStartingSeqNo() {
 	if o.mset == nil || o.mset.store == nil {
@@ -2656,17 +2590,6 @@ func (o *consumer) selectStartingSeqNo() {
 					ss := o.mset.store.FilteredState(1, o.cfg.FilterSubject)
 					o.sseq = ss.Last
 				}
-			} else if o.cfg.DeliverPolicy == DeliverLastPerSubject {
-				if mss := o.mset.store.SubjectsState(o.cfg.FilterSubject); len(mss) > 0 {
-					o.lss = &lastSeqSkipList{
-						resume: stats.LastSeq + 1,
-						seqs:   createLastSeqSkipList(mss),
-					}
-					o.sseq = o.lss.seqs[0]
-				} else {
-					// If no mapping info just set to last.
-					o.sseq = stats.LastSeq
-				}
 			} else if o.cfg.OptStartTime != nil {
 				// If we are here we are time based.
 				// TODO(dlc) - Once clustered can't rely on this.
@@ -3010,7 +2933,6 @@ func (o *consumer) setInitialPendingAndStart() {
 	if mset == nil || mset.store == nil {
 		return
 	}
-
 	// notFiltered means we want all messages.
 	notFiltered := o.cfg.FilterSubject == _EMPTY_
 	if !notFiltered {
@@ -3029,17 +2951,11 @@ func (o *consumer) setInitialPendingAndStart() {
 		}
 	} else {
 		// Here we are filtered.
-		dp := o.cfg.DeliverPolicy
-		if dp == DeliverLastPerSubject && o.hasSkipListPending() && o.sseq < o.lss.resume {
-			if o.lss != nil {
-				ss := mset.store.FilteredState(o.lss.resume, o.cfg.FilterSubject)
-				o.sseq = o.lss.seqs[0]
-				o.sgap = ss.Msgs + uint64(len(o.lss.seqs))
-			}
-		} else if ss := mset.store.FilteredState(o.sseq, o.cfg.FilterSubject); ss.Msgs > 0 {
+		ss := mset.store.FilteredState(o.sseq, o.cfg.FilterSubject)
+		if ss.Msgs > 0 {
 			o.sgap = ss.Msgs
 			// See if we should update our starting sequence.
-			if dp == DeliverLast || dp == DeliverLastPerSubject {
+			if dp := o.cfg.DeliverPolicy; dp == DeliverLast {
 				o.sseq = ss.Last
 			} else if dp == DeliverNew {
 				o.sseq = ss.Last + 1
@@ -3047,10 +2963,6 @@ func (o *consumer) setInitialPendingAndStart() {
 				// DeliverAll, DeliverByStartSequence, DeliverByStartTime
 				o.sseq = ss.First
 			}
-			// Cleanup lss when we take over in clustered mode.
-			if dp == DeliverLastPerSubject && o.hasSkipListPending() && o.sseq >= o.lss.resume {
-				o.lss = nil
-			}
 		}
 		o.updateSkipped()
 	}
diff --git a/server/filestore.go b/server/filestore.go
index d0578ee0..d9e81696 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -863,14 +863,122 @@ func (fs *fileStore) recoverMsgs() error {
 	fs.enforceBytesLimit()
 
 	// Do age checks too, make sure to call in place.
-	if fs.cfg.MaxAge != 0 && fs.state.Msgs > 0 {
+	if fs.cfg.MaxAge != 0 {
+		fs.expireMsgsOnRecover()
 		fs.startAgeChk()
-		fs.expireMsgsLocked()
 	}
 
 	return nil
 }
 
+// Will expire msgs that have aged out on restart.
+// We will treat this differently in case we have a recovery
+// that will expire alot of messages on startup. Should only be called
+// on startup. Lock should be held.
+func (fs *fileStore) expireMsgsOnRecover() {
+	if fs.state.Msgs == 0 {
+		return
+	}
+
+	var minAge = time.Now().UnixNano() - int64(fs.cfg.MaxAge)
+	var purged, bytes uint64
+	var deleted int
+
+	for _, mb := range fs.blks {
+		mb.mu.Lock()
+		if minAge < mb.first.ts {
+			mb.mu.Unlock()
+			break
+		}
+		// Can we remove whole block here?
+		if mb.last.ts <= minAge {
+			purged += mb.msgs
+			bytes += mb.bytes
+			mb.dirtyCloseWithRemove(true)
+			newFirst := mb.last.seq + 1
+			mb.mu.Unlock()
+			// Update fs first here as well.
+			fs.state.FirstSeq = newFirst
+			fs.state.FirstTime = time.Time{}
+			deleted++
+			continue
+		}
+
+		// If we are here we have to process the interior messages of this blk.
+		if err := mb.loadMsgsWithLock(); err != nil {
+			mb.mu.Unlock()
+			break
+		}
+
+		// Walk messages and remove if expired.
+		for seq := mb.first.seq; seq <= mb.last.seq; seq++ {
+			sm, err := mb.cacheLookupWithLock(seq)
+			// Process interior deleted msgs.
+			if err == errDeletedMsg {
+				// Update dmap.
+				if len(mb.dmap) > 0 {
+					delete(mb.dmap, seq)
+					if len(mb.dmap) == 0 {
+						mb.dmap = nil
+					}
+				}
+				continue
+			}
+			// Break on other errors.
+			if err != nil || sm == nil {
+				break
+			}
+
+			// No error and sm != nil from here onward.
+
+			// Check for done.
+			if sm.ts > minAge {
+				mb.first.seq = sm.seq
+				mb.first.ts = sm.ts
+				break
+			}
+
+			// Delete the message here.
+			sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
+			mb.bytes -= sz
+			bytes += sz
+			mb.msgs--
+			purged++
+			// Update fss
+			mb.removeSeqPerSubject(sm.subj, seq)
+		}
+
+		// Check if empty after processing, could happen if tail of messages are all deleted.
+		isEmpty := mb.msgs == 0
+		if isEmpty {
+			mb.dirtyCloseWithRemove(true)
+			// Update fs first here as well.
+			fs.state.FirstSeq = mb.last.seq + 1
+			fs.state.FirstTime = time.Time{}
+			deleted++
+		} else {
+			// Update fs first seq and time.
+			fs.state.FirstSeq = mb.first.seq
+			fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
+		}
+		mb.mu.Unlock()
+
+		if !isEmpty {
+			// Make sure to write out our index info.
+			mb.writeIndexInfo()
+		}
+		break
+	}
+
+	if deleted > 0 {
+		// Update blks slice.
+		fs.blks = append(fs.blks[:0:0], fs.blks[deleted:]...)
+	}
+	// Update top level accounting.
+	fs.state.Msgs -= purged
+	fs.state.Bytes -= bytes
+}
+
 // GetSeqFromTime looks for the first sequence number that has
 // the message with >= timestamp.
 // FIXME(dlc) - inefficient, and dumb really. Make this better.
@@ -1125,33 +1233,6 @@ func (fs *fileStore) perSubjectState(subj string) (total, first, last uint64) {
 	return total, first, last
 }
 
-// SubjectsState returns a map of SimpleState for all matching subjects.
-func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
-	fs.mu.RLock()
-	defer fs.mu.RUnlock()
-
-	if !fs.tms || fs.state.Msgs == 0 {
-		return nil
-	}
-
-	fss := make(map[string]SimpleState)
-	for _, mb := range fs.blks {
-		mb.mu.RLock()
-		for subj, ss := range mb.fss {
-			oss := fss[subj]
-			if oss.First == 0 { // New
-				fss[subj] = *ss
-			} else {
-				// Merge here.
-				oss.Last, oss.Msgs = ss.Last, oss.Msgs+ss.Msgs
-				fss[subj] = oss
-			}
-		}
-		mb.mu.RUnlock()
-	}
-	return fss
-}
-
 // RegisterStorageUpdates registers a callback for updates to storage changes.
 // It will present number of messages and bytes as a signed integer and an
 // optional sequence number of the message if a single.
@@ -2172,13 +2253,6 @@ func (fs *fileStore) startAgeChk() {
 	}
 }
 
-// Lock should be held.
-func (fs *fileStore) expireMsgsLocked() {
-	fs.mu.Unlock()
-	fs.expireMsgs()
-	fs.mu.Lock()
-}
-
 // Lock should be held.
 func (fs *fileStore) resetAgeChk(delta int64) {
 	fireIn := fs.cfg.MaxAge
@@ -3189,6 +3263,7 @@ func fileStoreMsgSizeEstimate(slen, maxPayload int) uint64 {
 }
 
 // Write index info to the appropriate file.
+// Lock should be held.
 func (mb *msgBlock) writeIndexInfo() error {
 	// HEADER: magic version msgs bytes fseq fts lseq lts ndel checksum
 	var hdr [indexHdrSize]byte
@@ -3586,8 +3661,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		return fs.purge(seq)
 	}
 
-	var purged uint64
-	var bytes uint64
+	var purged, bytes uint64
 
 	// We have to delete interior messages.
 	fs.mu.Lock()
@@ -3602,9 +3676,9 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 	}
 
 	// All msgblocks up to this one can be thrown away.
-	for i, mb := range fs.blks {
+	var deleted int
+	for _, mb := range fs.blks {
 		if mb == smb {
-			fs.blks = append(fs.blks[:0:0], fs.blks[i:]...)
 			break
 		}
 		mb.mu.Lock()
@@ -3612,35 +3686,62 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		bytes += mb.bytes
 		mb.dirtyCloseWithRemove(true)
 		mb.mu.Unlock()
+		deleted++
 	}
 
 	smb.mu.Lock()
 	for mseq := smb.first.seq; mseq < seq; mseq++ {
-		if sm, _ := smb.cacheLookupWithLock(mseq); sm != nil && smb.msgs > 0 {
+		sm, err := smb.cacheLookupWithLock(mseq)
+		if err == errDeletedMsg {
+			// Update dmap.
+			if len(smb.dmap) > 0 {
+				delete(smb.dmap, seq)
+				if len(smb.dmap) == 0 {
+					smb.dmap = nil
+				}
+			}
+		} else if sm != nil {
 			sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 			smb.bytes -= sz
 			bytes += sz
 			smb.msgs--
 			purged++
+			// Update fss
+			smb.removeSeqPerSubject(sm.subj, mseq)
 		}
 	}
 
-	// Update first entry.
-	sm, _ := smb.cacheLookupWithLock(seq)
-	if sm != nil {
-		smb.first.seq = sm.seq
-		smb.first.ts = sm.ts
+	// Check if empty after processing, could happen if tail of messages are all deleted.
+	isEmpty := smb.msgs == 0
+	if isEmpty {
+		smb.dirtyCloseWithRemove(true)
+		// Update fs first here as well.
+		fs.state.FirstSeq = smb.last.seq + 1
+		fs.state.FirstTime = time.Time{}
+		deleted++
+	} else {
+		// Update fs first seq and time.
+		smb.first.seq = seq - 1 // Just for start condition for selectNextFirst.
+		smb.selectNextFirst()
+		fs.state.FirstSeq = smb.first.seq
+		fs.state.FirstTime = time.Unix(0, smb.first.ts).UTC()
 	}
 	smb.mu.Unlock()
 
-	if sm != nil {
-		// Reset our version of first.
-		fs.state.FirstSeq = sm.seq
-		fs.state.FirstTime = time.Unix(0, sm.ts).UTC()
-		fs.state.Msgs -= purged
-		fs.state.Bytes -= bytes
+	if !isEmpty {
+		// Make sure to write out our index info.
+		smb.writeIndexInfo()
 	}
 
+	if deleted > 0 {
+		// Update blks slice.
+		fs.blks = append(fs.blks[:0:0], fs.blks[deleted:]...)
+	}
+
+	// Update top level accounting.
+	fs.state.Msgs -= purged
+	fs.state.Bytes -= bytes
+
 	cb := fs.scb
 	fs.mu.Unlock()
 
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 56c64b23..d0a55c84 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -3018,6 +3018,7 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, acc *Acco
 }
 
 func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subject, reply string, rmsg []byte, expectDurable bool) {
+
 	if c == nil || !s.JetStreamEnabled() {
 		return
 	}
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 9d371076..21c06150 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -3990,23 +3990,6 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 		return
 	}
 
-	// Also short circuit if DeliverLastPerSubject is set with no FilterSubject.
-	if cfg.DeliverPolicy == DeliverLastPerSubject {
-		badConfig := cfg.FilterSubject == _EMPTY_
-		if !badConfig {
-			subjects := sa.Config.Subjects
-			if len(subjects) == 1 && subjects[0] == cfg.FilterSubject && subjectIsLiteral(subjects[0]) {
-				badConfig = true
-			}
-		}
-		if badConfig {
-			resp.Error = ApiErrors[JSConsumerInvalidPolicyErrF].NewT("{err}",
-				"consumer delivery policy is deliver last per subject, but FilterSubject is not set")
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
-			return
-		}
-	}
-
 	// Setup proper default for ack wait if we are in explicit ack mode.
 	if cfg.AckWait == 0 && (cfg.AckPolicy == AckExplicit || cfg.AckPolicy == AckAll) {
 		cfg.AckWait = JsAckWaitDefault
diff --git a/server/memstore.go b/server/memstore.go
index d075d2ec..a84b5651 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -317,24 +317,6 @@ func (ms *memStore) filteredStateLocked(sseq uint64, subj string) SimpleState {
 	return ss
 }
 
-// SubjectsState returns a map of SimpleState for all matching subjects.
-func (ms *memStore) SubjectsState(subject string) map[string]SimpleState {
-	ms.mu.RLock()
-	defer ms.mu.RUnlock()
-
-	if len(ms.fss) == 0 {
-		return nil
-	}
-
-	fss := make(map[string]SimpleState)
-	for subj, ss := range ms.fss {
-		if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) {
-			fss[subj] = *ss
-		}
-	}
-	return fss
-}
-
 // Will check the msg limit for this tracked subject.
 // Lock should be held.
 func (ms *memStore) enforcePerSubjectLimit(ss *SimpleState) {
diff --git a/server/store.go b/server/store.go
index 5da8274a..2139d1a6 100644
--- a/server/store.go
+++ b/server/store.go
@@ -79,8 +79,7 @@ type StreamStore interface {
 	Compact(seq uint64) (uint64, error)
 	Truncate(seq uint64) error
 	GetSeqFromTime(t time.Time) uint64
-	FilteredState(seq uint64, subject string) SimpleState
-	SubjectsState(filterSubject string) map[string]SimpleState
+	FilteredState(sseq uint64, subject string) SimpleState
 	State() StreamState
 	FastState(*StreamState)
 	Type() StorageType
@@ -112,7 +111,7 @@ type DiscardPolicy int
 const (
 	// DiscardOld will remove older messages to return to the limits.
 	DiscardOld = iota
-	// DiscardNew will error on a StoreMsg call
+	//DiscardNew will error on a StoreMsg call
 	DiscardNew
 )
 
@@ -390,7 +389,6 @@ const (
 	deliverNewPolicyString       = "new"
 	deliverByStartSequenceString = "by_start_sequence"
 	deliverByStartTimeString     = "by_start_time"
-	deliverLastPerPolicyString   = "last_per_subject"
 	deliverUndefinedString       = "undefined"
 )
 
@@ -400,8 +398,6 @@ func (p *DeliverPolicy) UnmarshalJSON(data []byte) error {
 		*p = DeliverAll
 	case jsonString(deliverLastPolicyString):
 		*p = DeliverLast
-	case jsonString(deliverLastPerPolicyString):
-		*p = DeliverLastPerSubject
 	case jsonString(deliverNewPolicyString):
 		*p = DeliverNew
 	case jsonString(deliverByStartSequenceString):
@@ -421,8 +417,6 @@ func (p DeliverPolicy) MarshalJSON() ([]byte, error) {
 		return json.Marshal(deliverAllPolicyString)
 	case DeliverLast:
 		return json.Marshal(deliverLastPolicyString)
-	case DeliverLastPerSubject:
-		return json.Marshal(deliverLastPerPolicyString)
 	case DeliverNew:
 		return json.Marshal(deliverNewPolicyString)
 	case DeliverByStartSequence:


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 3f30ab50..5a59506f 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -566,7 +566,7 @@ func TestFileStoreBytesLimit(t *testing.T) {
 }
 
 func TestFileStoreAgeLimit(t *testing.T) {
-	maxAge := 100 * time.Millisecond
+	maxAge := 250 * time.Millisecond
 
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)
@@ -584,7 +584,9 @@ func TestFileStoreAgeLimit(t *testing.T) {
 	subj, msg := "foo", []byte("Hello World")
 	toStore := 500
 	for i := 0; i < toStore; i++ {
-		fs.StoreMsg(subj, nil, msg)
+		if _, _, err := fs.StoreMsg(subj, nil, msg); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 	}
 	state := fs.State()
 	if state.Msgs != uint64(toStore) {
@@ -614,7 +616,16 @@ func TestFileStoreAgeLimit(t *testing.T) {
 	if state.Msgs != uint64(toStore) {
 		t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 	}
+	fs.RemoveMsg(502)
+	fs.RemoveMsg(602)
+	fs.RemoveMsg(702)
+	fs.RemoveMsg(802)
+	// We will measure the time to make sure expires works with interior deletes.
+	start := time.Now()
 	checkExpired(t)
+	if elapsed := time.Since(start); elapsed > time.Second {
+		t.Fatalf("Took too long to expire: %v", elapsed)
+	}
 }
 
 func TestFileStoreTimeStamps(t *testing.T) {
@@ -2891,3 +2902,227 @@ func TestBadConsumerState(t *testing.T) {
 		t.Fatalf("Expected to not throw error, got %v and %+v", err, cs)
 	}
 }
+
+func TestFileStoreExpireMsgsOnStart(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	ttl := 250 * time.Millisecond
+	cfg := StreamConfig{Name: "ORDERS", Subjects: []string{"orders.*"}, Storage: FileStorage, MaxAge: ttl}
+	var fs *fileStore
+
+	startFS := func() *fileStore {
+		t.Helper()
+		fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 8 * 1024}, cfg)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		return fs
+	}
+
+	newFS := func() *fileStore {
+		t.Helper()
+		if fs != nil {
+			fs.Stop()
+			fs = nil
+		}
+		removeDir(t, storeDir)
+		return startFS()
+	}
+
+	restartFS := func(delay time.Duration) *fileStore {
+		if fs != nil {
+			fs.Stop()
+			fs = nil
+			time.Sleep(delay)
+		}
+		fs = startFS()
+		return fs
+	}
+
+	fs = newFS()
+	defer fs.Stop()
+
+	msg := bytes.Repeat([]byte("ABC"), 33) // ~100bytes
+	loadMsgs := func(n int) {
+		t.Helper()
+		for i := 1; i <= n; i++ {
+			if _, _, err := fs.StoreMsg(fmt.Sprintf("orders.%d", i%10), nil, msg); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+		}
+	}
+
+	checkState := func(msgs, first, last uint64) {
+		t.Helper()
+		if fs == nil {
+			t.Fatalf("No fs")
+			return
+		}
+		state := fs.State()
+		if state.Msgs != msgs {
+			t.Fatalf("Expected %d msgs, got %d", msgs, state.Msgs)
+		}
+		if state.FirstSeq != first {
+			t.Fatalf("Expected %d as first, got %d", first, state.FirstSeq)
+		}
+		if state.LastSeq != last {
+			t.Fatalf("Expected %d as last, got %d", last, state.LastSeq)
+		}
+	}
+
+	checkNumBlks := func(expected int) {
+		fs.mu.RLock()
+		n := len(fs.blks)
+		fs.mu.RUnlock()
+		if n != expected {
+			t.Fatalf("Expected %d msg blks, got %d", expected, n)
+		}
+	}
+
+	// Check the filtered subject state and make sure that is tracked properly.
+	checkFiltered := func(subject string, ss SimpleState) {
+		t.Helper()
+		fss := fs.FilteredState(1, subject)
+		if fss != ss {
+			t.Fatalf("Expected FilteredState of %+v, got %+v", ss, fss)
+		}
+	}
+
+	// Make sure state on disk matches (e.g. writeIndexInfo properly called)
+	checkBlkState := func(index int) {
+		t.Helper()
+		fs.mu.RLock()
+		if index >= len(fs.blks) {
+			t.Fatalf("Out of range, wanted %d but only %d blks", index, len(fs.blks))
+		}
+		mb := fs.blks[index]
+		fs.mu.RUnlock()
+
+		var errStr string
+
+		mb.mu.RLock()
+		// We will do a readIndex op on our clone and then compare.
+		mbc := &msgBlock{fs: fs, ifn: mb.ifn}
+		if err := mbc.readIndexInfo(); err != nil {
+			mb.mu.RUnlock()
+			t.Fatalf("Error during readIndexInfo: %v", err)
+		}
+		// Check state as represented by index info.
+		if mb.msgs != mbc.msgs {
+			errStr = fmt.Sprintf("msgs do not match: %d vs %d", mb.msgs, mbc.msgs)
+		} else if mb.bytes != mbc.bytes {
+			errStr = fmt.Sprintf("bytes do not match: %d vs %d", mb.bytes, mbc.bytes)
+		} else if mb.first != mbc.first {
+			errStr = fmt.Sprintf("first state does not match: %d vs %d", mb.first, mbc.first)
+		} else if mb.last != mbc.last {
+			errStr = fmt.Sprintf("last state does not match: %d vs %d", mb.last, mbc.last)
+		} else if !reflect.DeepEqual(mb.dmap, mbc.dmap) {
+			errStr = fmt.Sprintf("deleted map does not match: %d vs %d", mb.dmap, mbc.dmap)
+		}
+		mb.mu.RUnlock()
+		if errStr != _EMPTY_ {
+			t.Fatal(errStr)
+		}
+	}
+
+	lastSeqForBlk := func(index int) uint64 {
+		t.Helper()
+		fs.mu.RLock()
+		defer fs.mu.RUnlock()
+		if len(fs.blks) == 0 {
+			t.Fatalf("No blocks?")
+		}
+		mb := fs.blks[0]
+		mb.mu.RLock()
+		defer mb.mu.RUnlock()
+		return mb.last.seq
+	}
+
+	// Actual testing here.
+
+	loadMsgs(500)
+	restartFS(ttl + 100*time.Millisecond)
+	checkState(0, 501, 500)
+	checkNumBlks(0)
+
+	// Now check partial expires and the fss tracking state.
+	// Small numbers is to keep them in one block.
+	fs = newFS()
+	loadMsgs(10)
+	time.Sleep(100 * time.Millisecond)
+	loadMsgs(10)
+	restartFS(ttl - 100*time.Millisecond + 25*time.Millisecond) // Just want half
+	checkState(10, 11, 20)
+	checkNumBlks(1)
+	checkFiltered("orders.*", SimpleState{Msgs: 10, First: 11, Last: 20})
+	checkFiltered("orders.5", SimpleState{Msgs: 1, First: 15, Last: 15})
+	checkBlkState(0)
+
+	fs = newFS()
+	loadMsgs(5)
+	time.Sleep(100 * time.Millisecond)
+	loadMsgs(15)
+	restartFS(ttl - 100*time.Millisecond + 25*time.Millisecond) // Just want half
+	checkState(15, 6, 20)
+	checkFiltered("orders.*", SimpleState{Msgs: 15, First: 6, Last: 20})
+	checkFiltered("orders.5", SimpleState{Msgs: 2, First: 10, Last: 20})
+
+	// Now we want to test that if the end of a msg block is all deletes msgs that we do the right thing.
+	fs = newFS()
+	loadMsgs(150)
+	time.Sleep(100 * time.Millisecond)
+	loadMsgs(100)
+
+	checkNumBlks(5)
+
+	// Now delete 10 messages from the end of the first block which we will expire on restart.
+	// We will expire up to seq 100, so delete 91-100.
+	lseq := lastSeqForBlk(0)
+	for seq := lseq; seq > lseq-10; seq-- {
+		removed, err := fs.RemoveMsg(seq)
+		if err != nil || !removed {
+			t.Fatalf("Error removing message: %v", err)
+		}
+	}
+	restartFS(ttl - 100*time.Millisecond + 25*time.Millisecond) // Just want half
+	checkState(100, 151, 250)
+	checkNumBlks(3) // We should only have 3 blks left.
+	checkBlkState(0)
+
+	// Now make sure that we properly clean up any internal dmap entries (sparse) when expiring.
+	fs = newFS()
+	loadMsgs(10)
+	// Remove some in sparse fashion, adding to dmap.
+	fs.RemoveMsg(2)
+	fs.RemoveMsg(4)
+	fs.RemoveMsg(6)
+	time.Sleep(100 * time.Millisecond)
+	loadMsgs(10)
+	restartFS(ttl - 100*time.Millisecond + 25*time.Millisecond) // Just want half
+	checkState(10, 11, 20)
+	checkNumBlks(1)
+	checkBlkState(0)
+
+	// Make sure expiring a block with tail deleted messages removes the message block etc.
+	fs = newFS()
+	loadMsgs(7)
+	time.Sleep(100 * time.Millisecond)
+	loadMsgs(3)
+	fs.RemoveMsg(8)
+	fs.RemoveMsg(9)
+	fs.RemoveMsg(10)
+	restartFS(ttl - 100*time.Millisecond + 25*time.Millisecond)
+	checkState(0, 11, 10)
+
+	// Not for start per se but since we have all the test tooling here check that Compact() does right thing as well.
+	fs = newFS()
+	loadMsgs(100)
+	checkFiltered("orders.*", SimpleState{Msgs: 100, First: 1, Last: 100})
+	checkFiltered("orders.5", SimpleState{Msgs: 10, First: 5, Last: 95})
+	// Check that Compact keeps fss updated, does dmap etc.
+	fs.Compact(51)
+	checkFiltered("orders.*", SimpleState{Msgs: 50, First: 51, Last: 100})
+	checkFiltered("orders.5", SimpleState{Msgs: 5, First: 55, Last: 95})
+	checkBlkState(0)
+}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index f3c14343..73fe9dca 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -12169,148 +12169,6 @@ func TestJetStreamConsumerBadNumPending(t *testing.T) {
 	checkForNoPending(mon)
 }
 
-func TestJetStreamDeliverLastPerSubject(t *testing.T) {
-	for _, st := range []StorageType{FileStorage, MemoryStorage} {
-		t.Run(st.String(), func(t *testing.T) {
-			s := RunBasicJetStreamServer()
-			defer s.Shutdown()
-
-			if config := s.JetStreamConfig(); config != nil {
-				defer removeDir(t, config.StoreDir)
-			}
-
-			// Client for API requests.
-			nc, js := jsClientConnect(t, s)
-			defer nc.Close()
-
-			cfg := StreamConfig{
-				Name:       "KV",
-				Subjects:   []string{"kv.>"},
-				Storage:    st,
-				MaxMsgsPer: 5,
-			}
-
-			req, err := json.Marshal(cfg)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			// Do manually for now.
-			nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
-			si, err := js.StreamInfo("KV")
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			if si == nil || si.Config.Name != "KV" {
-				t.Fatalf("StreamInfo is not correct %+v", si)
-			}
-
-			// Interleave them on purpose.
-			for i := 1; i <= 11; i++ {
-				msg := []byte(fmt.Sprintf("%d", i))
-				js.PublishAsync("kv.b1.foo", msg)
-				js.PublishAsync("kv.b2.foo", msg)
-
-				js.PublishAsync("kv.b1.bar", msg)
-				js.PublishAsync("kv.b2.bar", msg)
-
-				js.PublishAsync("kv.b1.baz", msg)
-				js.PublishAsync("kv.b2.baz", msg)
-			}
-
-			select {
-			case <-js.PublishAsyncComplete():
-			case <-time.After(2 * time.Second):
-				t.Fatalf("Did not receive completion signal")
-			}
-
-			// Do quick check that config needs FilteredSubjects otherwise bad config.
-			badReq := CreateConsumerRequest{
-				Stream: "KV",
-				Config: ConsumerConfig{
-					DeliverSubject: "b",
-					DeliverPolicy:  DeliverLastPerSubject,
-				},
-			}
-			req, err = json.Marshal(badReq)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			resp, err := nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "KV"), req, time.Second)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			var ccResp JSApiConsumerCreateResponse
-			if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			if ccResp.Error == nil || !strings.Contains(ccResp.Error.Description, "filter subject is not set") {
-				t.Fatalf("Expected an error, got none")
-			}
-
-			// Now let's consume these via last per subject.
-			obsReq := CreateConsumerRequest{
-				Stream: "KV",
-				Config: ConsumerConfig{
-					DeliverSubject: "d",
-					DeliverPolicy:  DeliverLastPerSubject,
-					FilterSubject:  "kv.b1.*",
-				},
-			}
-			req, err = json.Marshal(obsReq)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "KV"), req, time.Second)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			ccResp.Error = nil
-			if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-
-			sub, _ := nc.SubscribeSync("d")
-			defer sub.Unsubscribe()
-
-			// Helper to check messages are correct.
-			checkNext := func(subject string, sseq uint64, v string) {
-				t.Helper()
-				m, err := sub.NextMsg(time.Second)
-				if err != nil {
-					t.Fatalf("Error receiving message: %v", err)
-				}
-				if m.Subject != subject {
-					t.Fatalf("Expected subject %q but got %q", subject, m.Subject)
-				}
-				meta, err := m.Metadata()
-				if err != nil {
-					t.Fatalf("didn't get metadata: %s", err)
-				}
-				if meta.Sequence.Stream != sseq {
-					t.Fatalf("Expected stream seq %d but got %d", sseq, meta.Sequence.Stream)
-				}
-				if string(m.Data) != v {
-					t.Fatalf("Expected data of %q but got %q", v, m.Data)
-				}
-			}
-
-			checkSubsPending(t, sub, 3)
-
-			// Now make sure they are what we expect.
-			checkNext("kv.b1.foo", 61, "11")
-			checkNext("kv.b1.bar", 63, "11")
-			checkNext("kv.b1.baz", 65, "11")
-
-			msg := []byte(fmt.Sprintf("%d", 22))
-			js.Publish("kv.b2.foo", msg) // Not filtered through..
-			js.Publish("kv.b1.bar", msg)
-
-			checkSubsPending(t, sub, 1)
-			checkNext("kv.b1.bar", 68, "22")
-		})
-	}
-}
-
 // We had a report of a consumer delete crashing the server when in interest retention mode.
 // This I believe is only really possible in clustered mode, but we will force the issue here.
 func TestJetStreamConsumerCleanupWithRetentionPolicy(t *testing.T) {
diff --git a/server/norace_test.go b/server/norace_test.go
index 9f0595f9..a948126b 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -24,6 +24,7 @@ import (
 	"net/url"
 	"runtime"
 	"runtime/debug"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -2411,3 +2412,77 @@ func TestNoRaceJetStreamFileStoreBufferReuse(t *testing.T) {
 	}
 	fmt.Printf("MEM AFTER SUBSCRIBE is %v\n", friendlyBytes(v.Mem))
 }
+
+// Report of slow restart for a server that has many messages that have expired while it was not running.
+func TestNoRaceJetStreamSlowRestartWithManyExpiredMsgs(t *testing.T) {
+	opts := DefaultTestOptions
+	opts.Port = -1
+	opts.JetStream = true
+	s := RunServer(&opts)
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	ttl := 2 * time.Second
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "ORDERS",
+		Subjects: []string{"orders.*"},
+		MaxAge:   ttl,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Attach a consumer who is filtering on a wildcard subject as well.
+	// This does not affect it like I thought originally but will keep it here.
+	_, err = js.AddConsumer("ORDERS", &nats.ConsumerConfig{
+		Durable:       "c22",
+		FilterSubject: "orders.*",
+		AckPolicy:     nats.AckExplicitPolicy,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Now fill up with messages.
+	toSend := 100_000
+	for i := 1; i <= toSend; i++ {
+		js.PublishAsync(fmt.Sprintf("orders.%d", i), []byte("OK"))
+	}
+	<-js.PublishAsyncComplete()
+
+	sdir := strings.TrimSuffix(s.JetStreamConfig().StoreDir, JetStreamStoreDir)
+	s.Shutdown()
+
+	// Let them expire while not running.
+	time.Sleep(ttl + 500*time.Millisecond)
+
+	start := time.Now()
+	opts.Port = -1
+	opts.StoreDir = sdir
+	s = RunServer(&opts)
+	elapsed := time.Since(start)
+	defer s.Shutdown()
+
+	if elapsed > 2*time.Second {
+		t.Fatalf("Took %v for restart which is too long", elapsed)
+	}
+
+	// Check everything is correct.
+	nc, js = jsClientConnect(t, s)
+	defer nc.Close()
+
+	si, err := js.StreamInfo("ORDERS")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if si.State.Msgs != 0 {
+		t.Fatalf("Expected no msgs after restart, got %d", si.State.Msgs)
+	}
+}