Repository: nats-io/nats-server
Pull request #1748 (2020-11-25 21:34:16)

Fixed bug restoring consumer state
This was also the flapper test we had for **TestJetStreamConsumerMaxDeliveryAndServerRestart**.
    
We were not properly restoring our state for consumers and we also had a bug where we would not properly encode and write redelivered state.
    
Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 605e15b1..2fcaf343 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -265,21 +265,12 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 
 	// Make sure any partition subject is also a literal.
 	if config.FilterSubject != "" {
-		// If this is a direct match for the streams only subject clear the filter.
-		mset.mu.RLock()
-		if len(mset.config.Subjects) == 1 && mset.config.Subjects[0] == config.FilterSubject {
-			config.FilterSubject = _EMPTY_
+		// Make sure this is a valid partition of the interest subjects.
+		if !mset.validSubject(config.FilterSubject) {
+			return nil, fmt.Errorf("consumer filter subject is not a valid subset of the interest subjects")
 		}
-		mset.mu.RUnlock()
-
-		if config.FilterSubject != "" {
-			// Make sure this is a valid partition of the interest subjects.
-			if !mset.validSubject(config.FilterSubject) {
-				return nil, fmt.Errorf("consumer filter subject is not a valid subset of the interest subjects")
-			}
-			if config.AckPolicy == AckAll {
-				return nil, fmt.Errorf("consumer with filter subject can not have an ack policy of ack all")
-			}
+		if config.AckPolicy == AckAll {
+			return nil, fmt.Errorf("consumer with filter subject can not have an ack policy of ack all")
 		}
 	}
 
@@ -762,15 +753,15 @@ func (o *Consumer) sendAckReply(subj string) {
 
 // Process a message for the ack reply subject delivered with a message.
 func (o *Consumer) processAck(_ *subscription, _ *client, subject, reply string, msg []byte) {
-	sseq, dseq, dcount := ackReplyInfo(subject)
+	sseq, dseq, dc := ackReplyInfo(subject)
 
 	skipAckReply := sseq == 0
 
 	switch {
 	case len(msg) == 0, bytes.Equal(msg, AckAck), bytes.Equal(msg, AckOK):
-		o.ackMsg(sseq, dseq, dcount)
+		o.ackMsg(sseq, dseq, dc)
 	case bytes.HasPrefix(msg, AckNext):
-		o.ackMsg(sseq, dseq, dcount)
+		o.ackMsg(sseq, dseq, dc)
 		o.processNextMsgReq(nil, nil, subject, reply, msg[len(AckNext):])
 		skipAckReply = true
 	case bytes.Equal(msg, AckNak):
@@ -778,7 +769,7 @@ func (o *Consumer) processAck(_ *subscription, _ *client, subject, reply string,
 	case bytes.Equal(msg, AckProgress):
 		o.progressUpdate(sseq)
 	case bytes.Equal(msg, AckTerm):
-		o.processTerm(sseq, dseq, dcount)
+		o.processTerm(sseq, dseq, dc)
 	}
 
 	// Ack the ack if requested.
@@ -824,9 +815,9 @@ func (o *Consumer) processNak(sseq, dseq uint64) {
 }
 
 // Process a TERM
-func (o *Consumer) processTerm(sseq, dseq, dcount uint64) {
+func (o *Consumer) processTerm(sseq, dseq, dc uint64) {
 	// Treat like an ack to suppress redelivery.
-	o.processAckMsg(sseq, dseq, dcount, false)
+	o.processAckMsg(sseq, dseq, dc, false)
 
 	o.mu.Lock()
 	defer o.mu.Unlock()
@@ -842,7 +833,7 @@ func (o *Consumer) processTerm(sseq, dseq, dcount uint64) {
 		Consumer:    o.name,
 		ConsumerSeq: dseq,
 		StreamSeq:   sseq,
-		Deliveries:  dcount,
+		Deliveries:  dc,
 	}
 
 	j, err := json.MarshalIndent(e, "", "  ")
@@ -1031,7 +1022,7 @@ func (o *Consumer) shouldSample() bool {
 	return mrand.Int31n(100) <= o.sfreq
 }
 
-func (o *Consumer) sampleAck(sseq, dseq, dcount uint64) {
+func (o *Consumer) sampleAck(sseq, dseq, dc uint64) {
 	if !o.shouldSample() {
 		return
 	}
@@ -1050,7 +1041,7 @@ func (o *Consumer) sampleAck(sseq, dseq, dcount uint64) {
 		ConsumerSeq: dseq,
 		StreamSeq:   sseq,
 		Delay:       unow - o.pending[sseq].Timestamp,
-		Deliveries:  dcount,
+		Deliveries:  dc,
 	}
 
 	j, err := json.MarshalIndent(e, "", "  ")
@@ -1062,11 +1053,11 @@ func (o *Consumer) sampleAck(sseq, dseq, dcount uint64) {
 }
 
 // Process an ack for a message.
-func (o *Consumer) ackMsg(sseq, dseq, dcount uint64) {
-	o.processAckMsg(sseq, dseq, dcount, true)
+func (o *Consumer) ackMsg(sseq, dseq, dc uint64) {
+	o.processAckMsg(sseq, dseq, dc, true)
 }
 
-func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
+func (o *Consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 	o.mu.Lock()
 	var sagap uint64
 	var needSignal bool
@@ -1075,7 +1066,7 @@ func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
 	case AckExplicit:
 		if p, ok := o.pending[sseq]; ok {
 			if doSample {
-				o.sampleAck(sseq, dseq, dcount)
+				o.sampleAck(sseq, dseq, dc)
 			}
 			if o.maxp > 0 && len(o.pending) >= o.maxp {
 				needSignal = true
@@ -1363,7 +1354,7 @@ func (o *Consumer) incDeliveryCount(sseq uint64) uint64 {
 }
 
 // send a delivery exceeded advisory.
-func (o *Consumer) notifyDeliveryExceeded(sseq, dcount uint64) {
+func (o *Consumer) notifyDeliveryExceeded(sseq, dc uint64) {
 	e := JSConsumerDeliveryExceededAdvisory{
 		TypedEvent: TypedEvent{
 			Type: JSConsumerDeliveryExceededAdvisoryType,
@@ -1373,7 +1364,7 @@ func (o *Consumer) notifyDeliveryExceeded(sseq, dcount uint64) {
 		Stream:     o.stream,
 		Consumer:   o.name,
 		StreamSeq:  sseq,
-		Deliveries: dcount,
+		Deliveries: dc,
 	}
 
 	j, err := json.MarshalIndent(e, "", "  ")
@@ -1405,20 +1396,20 @@ var errBadConsumer = errors.New("consumer not valid")
 // Get next available message from underlying store.
 // Is partition aware and redeliver aware.
 // Lock should be held.
-func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dcount uint64, ts int64, err error) {
+func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dc uint64, ts int64, err error) {
 	if o.mset == nil {
 		return _EMPTY_, nil, nil, 0, 0, 0, errBadConsumer
 	}
 	for {
-		seq, dcount := o.sseq, uint64(1)
+		seq, dc := o.sseq, uint64(1)
 		if len(o.rdq) > 0 {
 			seq = o.rdq[0]
 			o.rdq = append(o.rdq[:0], o.rdq[1:]...)
-			dcount = o.incDeliveryCount(seq)
-			if o.maxdc > 0 && dcount > o.maxdc {
+			dc = o.incDeliveryCount(seq)
+			if o.maxdc > 0 && dc > o.maxdc {
 				// Only send once
-				if dcount == o.maxdc+1 {
-					o.notifyDeliveryExceeded(seq, dcount-1)
+				if dc == o.maxdc+1 {
+					o.notifyDeliveryExceeded(seq, dc-1)
 				}
 				// Make sure to remove from pending.
 				delete(o.pending, seq)
@@ -1432,14 +1423,14 @@ func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dcoun
 
 		subj, hdr, msg, ts, err := o.mset.store.LoadMsg(seq)
 		if err == nil {
-			if dcount == 1 { // First delivery.
+			if dc == 1 { // First delivery.
 				o.sseq++
 				if o.config.FilterSubject != _EMPTY_ && !o.isFilteredMatch(subj) {
 					continue
 				}
 			}
 			// We have the msg here.
-			return subj, hdr, msg, seq, dcount, ts, nil
+			return subj, hdr, msg, seq, dc, ts, nil
 		}
 		// We got an error here. If this is an EOF we will return, otherwise
 		// we can continue looking.
@@ -1517,7 +1508,7 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 	// Deliver all the msgs we have now, once done or on a condition, we wait for new ones.
 	for {
 		var (
-			seq, dcount uint64
+			seq, dc     uint64
 			subj, dsubj string
 			hdr         []byte
 			msg         []byte
@@ -1543,7 +1534,7 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 			goto waitForMsgs
 		}
 
-		subj, hdr, msg, seq, dcount, ts, err = o.getNextMsg()
+		subj, hdr, msg, seq, dc, ts, err = o.getNextMsg()
 
 		// On error either wait or return.
 		if err != nil {
@@ -1595,7 +1586,7 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 			}
 		}
 
-		o.deliverMsg(dsubj, subj, hdr, msg, seq, dcount, ts)
+		o.deliverMsg(dsubj, subj, hdr, msg, seq, dc, ts)
 
 		o.mu.Unlock()
 		continue
@@ -1619,8 +1610,8 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 	}
 }
 
-func (o *Consumer) ackReply(sseq, dseq, dcount uint64, ts int64, pending uint64) string {
-	return fmt.Sprintf(o.ackReplyT, dcount, sseq, dseq, ts, pending)
+func (o *Consumer) ackReply(sseq, dseq, dc uint64, ts int64, pending uint64) string {
+	return fmt.Sprintf(o.ackReplyT, dc, sseq, dseq, ts, pending)
 }
 
 // deliverCurrentMsg is the hot path to deliver a message that was just received.
@@ -1676,12 +1667,12 @@ func (o *Consumer) deliverCurrentMsg(subj string, hdr, msg []byte, seq uint64, t
 
 // Deliver a msg to the consumer.
 // Lock should be held and o.mset validated to be non-nil.
-func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount uint64, ts int64) {
+func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint64, ts int64) {
 	if o.mset == nil {
 		return
 	}
 	// Update pending on first attempt
-	if dcount == 1 && o.sgap > 0 {
+	if dc == 1 && o.sgap > 0 {
 		o.sgap--
 	}
 
@@ -1693,7 +1684,7 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 	}
 
 	dseq := o.dseq
-	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dcount, ts, o.sgap), hdr, msg, o, seq}
+	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dc, ts, o.sgap), hdr, msg, o, seq}
 	mset := o.mset
 	ap := o.config.AckPolicy
 
@@ -1718,7 +1709,7 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 	o.dseq++
 
 	// FIXME(dlc) - Capture errors?
-	o.store.UpdateDelivered(dseq, seq, dcount, ts)
+	o.store.UpdateDelivered(dseq, seq, dc, ts)
 }
 
 // Tracks our outstanding pending acks. Only applicable to AckExplicit mode.
@@ -1868,7 +1859,7 @@ func parseAckReplyNum(d string) (n int64) {
 const expectedNumReplyTokens = 9
 
 // Grab encoded information in the reply subject for a delivered message.
-func (o *Consumer) ReplyInfo(subject string) (sseq, dseq, dcount uint64, ts int64, pending uint64) {
+func (o *Consumer) ReplyInfo(subject string) (sseq, dseq, dc uint64, ts int64, pending uint64) {
 	tsa := [expectedNumReplyTokens]string{}
 	start, tokens := 0, tsa[:0]
 	for i := 0; i < len(subject); i++ {
@@ -1883,15 +1874,15 @@ func (o *Consumer) ReplyInfo(subject string) (sseq, dseq, dcount uint64, ts int6
 	}
 	// TODO(dlc) - Should we error if we do not match consumer name?
 	// stream is tokens[2], consumer is 3.
-	dcount = uint64(parseAckReplyNum(tokens[4]))
+	dc = uint64(parseAckReplyNum(tokens[4]))
 	sseq, dseq = uint64(parseAckReplyNum(tokens[5])), uint64(parseAckReplyNum(tokens[6]))
 	ts = parseAckReplyNum(tokens[7])
 	pending = uint64(parseAckReplyNum(tokens[8]))
 
-	return sseq, dseq, dcount, ts, pending
+	return sseq, dseq, dc, ts, pending
 }
 
-func ackReplyInfo(subject string) (sseq, dseq, dcount uint64) {
+func ackReplyInfo(subject string) (sseq, dseq, dc uint64) {
 	tsa := [expectedNumReplyTokens]string{}
 	start, tokens := 0, tsa[:0]
 	for i := 0; i < len(subject); i++ {
@@ -1904,10 +1895,10 @@ func ackReplyInfo(subject string) (sseq, dseq, dcount uint64) {
 	if len(tokens) != expectedNumReplyTokens || tokens[0] != "$JS" || tokens[1] != "ACK" {
 		return 0, 0, 0
 	}
-	dcount = uint64(parseAckReplyNum(tokens[4]))
+	dc = uint64(parseAckReplyNum(tokens[4]))
 	sseq, dseq = uint64(parseAckReplyNum(tokens[5])), uint64(parseAckReplyNum(tokens[6]))
 
-	return sseq, dseq, dcount
+	return sseq, dseq, dc
 }
 
 // NextSeq returns the next delivered sequence number for this observable.
diff --git a/server/filestore.go b/server/filestore.go
index fd1520fa..3f4a0a86 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2897,11 +2897,12 @@ func (o *consumerFileStore) UpdateDelivered(dseq, sseq, dc uint64, ts int64) err
 			o.state.Delivered.Consumer = dseq
 			o.state.Delivered.Stream = sseq
 			p = &Pending{dseq, ts}
-		} else if dc > 1 {
+		}
+		if dc > 1 {
 			if o.state.Redelivered == nil {
 				o.state.Redelivered = make(map[uint64]uint64)
 			}
-			o.state.Redelivered[sseq] = dc
+			o.state.Redelivered[sseq] = dc - 1
 		}
 		o.state.Pending[sseq] = &Pending{dseq, ts}
 	} else {
@@ -2932,27 +2933,33 @@ func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
 	if p == nil {
 		return ErrStoreMsgNotFound
 	}
+	// Delete from our state.
 	delete(o.state.Pending, sseq)
+	if len(o.state.Redelivered) > 0 {
+		delete(o.state.Redelivered, sseq)
+		if len(o.state.Redelivered) == 0 {
+			o.state.Redelivered = nil
+		}
+	}
 
-	// TODO(dlc) - Check to see if we move ack floor.
 	if len(o.state.Pending) == 0 {
 		o.state.Pending = nil
 		o.state.AckFloor.Consumer = o.state.Delivered.Consumer
 		o.state.AckFloor.Stream = o.state.Delivered.Stream
-	} else if o.state.AckFloor.Consumer == 0 {
-		o.state.AckFloor.Consumer = dseq
-		o.state.AckFloor.Stream = sseq
 	} else if o.state.AckFloor.Consumer == dseq-1 {
+		notFirst := o.state.AckFloor.Consumer != 0
 		o.state.AckFloor.Consumer = dseq
 		o.state.AckFloor.Stream = sseq
-		// Close gap if needed.
-		for ss := sseq + 1; ss < o.state.Delivered.Stream; ss++ {
-			if p, ok := o.state.Pending[ss]; ok {
-				if p.Sequence > 0 {
-					o.state.AckFloor.Consumer = p.Sequence - 1
-					o.state.AckFloor.Stream = ss - 1
+		// Close the gap if needed.
+		if notFirst && o.state.Delivered.Consumer > dseq {
+			for ss := sseq + 1; ss < o.state.Delivered.Stream; ss++ {
+				if p, ok := o.state.Pending[ss]; ok {
+					if p.Sequence > 0 {
+						o.state.AckFloor.Consumer = p.Sequence - 1
+						o.state.AckFloor.Stream = ss - 1
+					}
+					break
 				}
-				break
 			}
 		}
 	}
@@ -3028,6 +3035,7 @@ func (o *consumerFileStore) encodeState() ([]byte, error) {
 			n += binary.PutUvarint(buf[n:], v)
 		}
 	}
+
 	return buf[:n], nil
 }
 
@@ -3323,6 +3331,23 @@ func (o *consumerFileStore) State() (*ConsumerState, error) {
 			state.Redelivered[seq] = n
 		}
 	}
+
+	// Copy this state into our own.
+	o.state.Delivered = state.Delivered
+	o.state.AckFloor = state.AckFloor
+	if len(state.Pending) > 0 {
+		o.state.Pending = make(map[uint64]*Pending, len(state.Pending))
+		for seq, p := range state.Pending {
+			o.state.Pending[seq] = &Pending{p.Sequence, p.Timestamp}
+		}
+	}
+	if len(state.Redelivered) > 0 {
+		o.state.Redelivered = make(map[uint64]uint64, len(state.Redelivered))
+		for seq, dc := range state.Redelivered {
+			o.state.Redelivered[seq] = dc
+		}
+	}
+
 	return state, nil
 }
 
diff --git a/server/jetstream.go b/server/jetstream.go
index 88f62a72..16959c85 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -607,35 +607,18 @@ func (a *Account) NumStreams() int {
 
 // Streams will return all known streams.
 func (a *Account) Streams() []*Stream {
-	return a.filteredStreams(_EMPTY_)
-}
-
-func (a *Account) filteredStreams(filter string) []*Stream {
 	a.mu.RLock()
 	jsa := a.js
 	a.mu.RUnlock()
-
 	if jsa == nil {
 		return nil
 	}
-
-	jsa.mu.Lock()
-	defer jsa.mu.Unlock()
-
 	var msets []*Stream
+	jsa.mu.Lock()
 	for _, mset := range jsa.streams {
-		if filter != _EMPTY_ {
-			for _, subj := range mset.config.Subjects {
-				if SubjectsCollide(filter, subj) {
-					msets = append(msets, mset)
-					break
-				}
-			}
-		} else {
-			msets = append(msets, mset)
-		}
+		msets = append(msets, mset)
 	}
-
+	jsa.mu.Unlock()
 	return msets
 }
 
@@ -657,6 +640,33 @@ func (a *Account) LookupStream(name string) (*Stream, error) {
 	return mset, nil
 }
 
+// LookupStreamBySubject will lookup a stream by a subject.
+// The subject needs to be an exact match or a subset. Supersets will not match.
+func (a *Account) LookupStreamBySubject(subj string) (*Stream, bool, error) {
+	a.mu.RLock()
+	jsa := a.js
+	a.mu.RUnlock()
+
+	if jsa == nil {
+		return nil, false, ErrJetStreamNotEnabled
+	}
+	jsa.mu.Lock()
+	defer jsa.mu.Unlock()
+
+	for _, mset := range jsa.streams {
+		for _, tsubj := range mset.config.Subjects {
+			if subj == tsubj {
+				return mset, len(mset.config.Subjects) > 1, nil
+			}
+			if subjectIsSubsetMatch(subj, tsubj) {
+				return mset, true, nil
+			}
+		}
+	}
+
+	return nil, false, ErrJetStreamStreamNotFound
+}
+
 // UpdateJetStreamLimits will update the account limits for a JetStream enabled account.
 func (a *Account) UpdateJetStreamLimits(limits *JetStreamAccountLimits) error {
 	a.mu.RLock()
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 48ce47a6..8b34cb30 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -75,6 +75,10 @@ const (
 	JSApiStreamInfo  = "$JS.API.STREAM.INFO.*"
 	JSApiStreamInfoT = "$JS.API.STREAM.INFO.%s"
 
+	// JSApiStreamLookup is for obtaining a stream by a target subject.
+	// Will return JSON response.
+	JSApiStreamLookup = "$JS.API.STREAM.LOOKUP"
+
 	// JSApiStreamDelete is the endpoint to delete streams.
 	// Will return JSON response.
 	JSApiStreamDelete  = "$JS.API.STREAM.DELETE.*"
@@ -143,8 +147,13 @@ const (
 	jsSnapshotAckT    = "$JS.SNAPSHOT.ACK.%s.%s"
 	jsRestoreDeliverT = "$JS.SNAPSHOT.RESTORE.%s.%s"
 
+	///////////////////////
+	// FIXME(dlc)
+	///////////////////////
+
 	// JetStreamAckT is the template for the ack message stream coming back from a consumer
 	// when they ACK/NAK, etc a message.
+	// FIXME(dlc) - What do we really need here??
 	jsAckT   = "$JS.ACK.%s.%s"
 	jsAckPre = "$JS.ACK."
 
@@ -257,6 +266,20 @@ type JSApiStreamInfoResponse struct {
 
 const JSApiStreamInfoResponseType = "io.nats.jetstream.api.v1.stream_info_response"
 
+type JSApiStreamLookupRequest struct {
+	// Subject finds any stream that matches this subject including those where wildcards intercepts
+	Subject string `json:"subject"`
+}
+
+// JSApiStreamLookupResponse.
+type JSApiStreamLookupResponse struct {
+	ApiResponse
+	Stream   string `json:"stream"`
+	Filtered bool   `json:"is_filtered"`
+}
+
+const JSApiStreamLookupResponseType = "io.nats.jetstream.api.v1.stream_lookup_response"
+
 // Maximum entries we will return for streams or consumers lists.
 // TODO(dlc) - with header or request support could request chunked response.
 const JSApiNamesLimit = 1024
@@ -264,8 +287,6 @@ const JSApiListLimit = 256
 
 type JSApiStreamNamesRequest struct {
 	ApiPagedRequest
-	// These are filters that can be applied to the list.
-	Subject string `json:"subject,omitempty"`
 }
 
 // JSApiStreamNamesResponse list of streams.
@@ -477,6 +498,7 @@ var allJsExports = []string{
 	JSApiStreams,
 	JSApiStreamList,
 	JSApiStreamInfo,
+	JSApiStreamLookup,
 	JSApiStreamDelete,
 	JSApiStreamPurge,
 	JSApiStreamSnapshot,
@@ -506,6 +528,7 @@ func (s *Server) setJetStreamExportSubs() error {
 		{JSApiStreams, s.jsStreamNamesRequest},
 		{JSApiStreamList, s.jsStreamListRequest},
 		{JSApiStreamInfo, s.jsStreamInfoRequest},
+		{JSApiStreamLookup, s.jsStreamLookupRequest},
 		{JSApiStreamDelete, s.jsStreamDeleteRequest},
 		{JSApiStreamPurge, s.jsStreamPurgeRequest},
 		{JSApiStreamSnapshot, s.jsStreamSnapshotRequest},
@@ -835,8 +858,6 @@ func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, subject, rep
 	}
 
 	var offset int
-	var filter string
-
 	if !isEmptyRequest(msg) {
 		var req JSApiStreamNamesRequest
 		if err := json.Unmarshal(msg, &req); err != nil {
@@ -845,15 +866,11 @@ func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, subject, rep
 			return
 		}
 		offset = req.Offset
-		if req.Subject != _EMPTY_ {
-			filter = req.Subject
-		}
 	}
 
 	// TODO(dlc) - Maybe hold these results for large results that we expect to be paged.
 	// TODO(dlc) - If this list is long maybe do this in a Go routine?
-	msets := c.acc.filteredStreams(filter)
-	// Since we page results order matters.
+	msets := c.acc.Streams()
 	sort.Slice(msets, func(i, j int) bool {
 		return strings.Compare(msets[i].config.Name, msets[j].config.Name) < 0
 	})
@@ -956,6 +973,57 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 	s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
+// Request to lookup a stream by target subject.
+func (s *Server) jsStreamLookupRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
+	if c == nil || c.acc == nil {
+		return
+	}
+
+	var resp = JSApiStreamLookupResponse{ApiResponse: ApiResponse{Type: JSApiStreamLookupResponseType}}
+	if !c.acc.JetStreamEnabled() {
+		resp.Error = jsNotEnabledErr
+		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	if isEmptyRequest(msg) {
+		resp.Error = &ApiError{Code: 400, Description: "subject required"}
+		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	subj := string(msg)
+
+	if bytes.HasPrefix(bytes.TrimSpace(msg), []byte("{")) {
+		var req JSApiStreamLookupRequest
+		err := json.Unmarshal(msg, &req)
+		// happy path we set it, otherwise we try whatever was in the
+		// msg as if its a subject this is because '{' is a valid subject
+		if err == nil {
+			subj = req.Subject
+		}
+	}
+
+	if !IsValidSubject(subj) {
+		resp.Error = &ApiError{Code: 400, Description: "subject argument is not a valid subject or request"}
+		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	// Lookup our stream.
+	mset, filtered, err := c.acc.LookupStreamBySubject(subj)
+	if err != nil {
+		resp.Error = jsNotFoundError(err)
+		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	resp.Stream = mset.Name()
+	resp.Filtered = filtered
+
+	s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(resp))
+}
+
 func isEmptyRequest(req []byte) bool {
 	if len(req) == 0 {
 		return true
diff --git a/server/stream.go b/server/stream.go
index 014b640a..60a72178 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -56,7 +56,7 @@ const JSApiPubAckResponseType = "io.nats.jetstream.api.v1.pub_ack_response"
 
 // JSPubAckResponse is a formal response to a publish operation.
 type JSPubAckResponse struct {
-	Error *ApiError `json:"error,omitempty"`
+	ApiResponse
 	*PubAck
 }
 
@@ -196,7 +196,10 @@ func (a *Account) AddStreamWithStore(config *StreamConfig, fsConfig *FileStoreCo
 	mset.setupSendCapabilities()
 
 	// Create our pubAck template here. Better than json marshal each time on success.
-	b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: cfg.Name, Sequence: math.MaxUint64}})
+	b, _ := json.Marshal(&JSPubAckResponse{
+		ApiResponse: ApiResponse{Type: JSApiPubAckResponseType},
+		PubAck:      &PubAck{Stream: cfg.Name, Sequence: math.MaxUint64},
+	})
 	end := bytes.Index(b, []byte(strconv.FormatUint(math.MaxUint64, 10)))
 	// We need to force cap here to make sure this is a copy when sending a response.
 	mset.pubAck = b[:end:end]
@@ -919,7 +922,7 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 	numConsumers := len(mset.consumers)
 	interestRetention := mset.config.Retention == InterestPolicy
 
-	var resp = &JSPubAckResponse{}
+	var resp = &JSPubAckResponse{ApiResponse: ApiResponse{Type: JSApiPubAckResponseType}}
 
 	// Process msg headers if present.
 	var msgId string
diff --git a/server/sublist.go b/server/sublist.go
index 1c860a83..37ea486e 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -1117,10 +1117,6 @@ func SubjectsCollide(subj1, subj2 string) bool {
 		return isSubsetMatch(toks2, subj1)
 	}
 	// Both have wildcards.
-	// If they only have partials then the lengths must match.
-	if !fwc1 && !fwc2 && len(toks1) != len(toks2) {
-		return false
-	}
 	stop := len(toks1)
 	if len(toks2) < stop {
 		stop = len(toks2)


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index e9bcb935..4bdc2c01 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -2026,6 +2026,74 @@ func TestFileStorePubPerfWithSmallBlkSize(t *testing.T) {
 	fmt.Printf("%s per sec\n", FriendlyBytes(int64(float64(toStore*storedMsgSize)/tt.Seconds())))
 }
 
+// Saw this manifest from a restart test with max delivered set for JetStream.
+func TestFileStoreConsumerRedeliveredLost(t *testing.T) {
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	cfg := &ConsumerConfig{AckPolicy: AckExplicit}
+	o, err := fs.ConsumerStore("o22", cfg)
+	if err != nil {
+		t.Fatalf("Unexepected error: %v", err)
+	}
+
+	restartConsumer := func() {
+		t.Helper()
+		o.Stop()
+		o, err = fs.ConsumerStore("o22", cfg)
+		if err != nil {
+			t.Fatalf("Unexepected error: %v", err)
+		}
+		// Make sure we recovered Redelivered.
+		state, err := o.State()
+		if err != nil {
+			t.Fatalf("Unexepected error: %v", err)
+		}
+		if len(state.Redelivered) == 0 {
+			t.Fatalf("Did not recover redelivered")
+		}
+	}
+
+	ts := time.Now().UnixNano()
+	o.UpdateDelivered(1, 1, 1, ts)
+	o.UpdateDelivered(2, 1, 2, ts)
+	o.UpdateDelivered(3, 1, 3, ts)
+	o.UpdateDelivered(4, 1, 4, ts)
+	o.UpdateDelivered(5, 2, 1, ts)
+
+	restartConsumer()
+
+	o.UpdateDelivered(6, 2, 2, ts)
+	o.UpdateDelivered(7, 3, 1, ts)
+
+	restartConsumer()
+	if state, _ := o.State(); len(state.Pending) != 3 {
+		t.Fatalf("Did not recover pending correctly")
+	}
+
+	o.UpdateAcks(7, 3)
+	o.UpdateAcks(6, 2)
+
+	restartConsumer()
+	o.UpdateAcks(4, 1)
+
+	state, _ := o.State()
+	if len(state.Pending) != 0 {
+		t.Fatalf("Did not clear pending correctly")
+	}
+	if len(state.Redelivered) != 0 {
+		fmt.Printf("redelivered is %+v\n", state.Redelivered)
+		t.Fatalf("Did not clear redelivered correctly")
+	}
+}
+
 func TestFileStoreConsumerFlusher(t *testing.T) {
 	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
 	os.MkdirAll(storeDir, 0755)
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index 580df444..f9090d7d 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -401,6 +401,109 @@ func TestJetStreamPubAck(t *testing.T) {
 	}
 }
 
+func TestJetStreamLookupStreamBySubject(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+	acc := s.GlobalAccount()
+
+	if _, err := acc.AddStream(&server.StreamConfig{Name: "1", Subjects: []string{"foo"}}); err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+	if _, err := acc.AddStream(&server.StreamConfig{Name: "2", Subjects: []string{"bar", "baz", "boo"}}); err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+	if _, err := acc.AddStream(&server.StreamConfig{Name: "3", Subjects: []string{"foo.*", "bar.*"}}); err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+	if _, err := acc.AddStream(&server.StreamConfig{Name: "4", Subjects: []string{"baz.*.*.>"}}); err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+	if _, err := acc.AddStream(&server.StreamConfig{Name: "5", Subjects: []string{"{test"}}); err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+
+	// Check some errors first.
+	checkError := func(subj string) {
+		t.Helper()
+		if _, _, err := acc.LookupStreamBySubject(subj); err != server.ErrJetStreamStreamNotFound {
+			t.Fatalf("Expected to get a stream not found error, got %v", err)
+		}
+	}
+
+	checkError("zzz")
+	checkError("*")
+	checkError("baz.>")
+
+	checkLookup := func(subj, stream string, filtered bool) {
+		t.Helper()
+		s, f, err := acc.LookupStreamBySubject(subj)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if s.Name() != stream {
+			t.Fatalf("Expected stream name of %q, got %q", stream, s.Name())
+		}
+		if f != filtered {
+			t.Fatalf("Expected filtered to be %v, got %v", filtered, f)
+		}
+	}
+
+	checkLookup("foo", "1", false)
+	checkLookup("boo", "2", true)
+	checkLookup("foo.*", "3", true)
+	checkLookup("foo.1", "3", true)
+	checkLookup("baz.*.*.>", "4", false)
+	checkLookup("baz.2.*.>", "4", true)
+
+	// Now test API
+	nc := clientConnectToServer(t, s)
+	defer nc.Close()
+
+	checkAPILookup := func(subj, stream string, filtered bool) {
+		t.Helper()
+		resp, err := nc.Request(server.JSApiStreamLookup, []byte(subj), time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var lresp server.JSApiStreamLookupResponse
+		if err := json.Unmarshal(resp.Data, &lresp); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if lresp.Error != nil {
+			t.Fatalf("Got an API error: %+v", lresp.Error)
+		}
+		if lresp.Stream != stream {
+			t.Fatalf("Expected stream name of %q, got %q", stream, lresp.Stream)
+		}
+		if lresp.Filtered != filtered {
+			t.Fatalf("Expected filtered to be %v, got %v", filtered, lresp.Filtered)
+		}
+	}
+
+	cases := []struct {
+		subj     string
+		stream   string
+		filtered bool
+	}{
+		{"foo", "1", false},
+		{"boo", "2", true},
+		{"foo.*", "3", true},
+		{"foo.1", "3", true},
+		{"baz.*.*.>", "4", false},
+		{"baz.2.*.>", "4", true},
+	}
+	for _, c := range cases {
+		checkAPILookup(c.subj, c.stream, c.filtered)
+		checkAPILookup(fmt.Sprintf(`{"subject":%q}`, c.subj), c.stream, c.filtered)
+	}
+
+	checkAPILookup("{test", "5", false)
+}
+
 func TestJetStreamConsumerWithStartTime(t *testing.T) {
 	subj := "my_stream"
 	cases := []struct {
@@ -7069,7 +7172,6 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	if err = json.Unmarshal(resp.Data, &namesResponse); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-
 	if len(namesResponse.Streams) != 1 {
 		t.Fatalf("Expected only 1 stream but got %d", len(namesResponse.Streams))
 	}
@@ -7493,59 +7595,6 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	checkEmptyReqArg(" { } ")
 }
 
-func TestJetStreamFilteredStreamNames(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer s.Shutdown()
-
-	// Forced cleanup of all persisted state.
-	if config := s.JetStreamConfig(); config != nil {
-		defer os.RemoveAll(config.StoreDir)
-	}
-
-	// Client for API requests.
-	nc := clientConnectToServer(t, s)
-	defer nc.Close()
-
-	// Create some streams.
-	var snid int
-	createStream := func(subjects []string) {
-		t.Helper()
-		snid++
-		name := fmt.Sprintf("S-%d", snid)
-		sc := &server.StreamConfig{Name: name, Subjects: subjects}
-		if _, err := s.GlobalAccount().AddStream(sc); err != nil {
-			t.Fatalf("Unexpected error adding stream: %v", err)
-		}
-	}
-
-	createStream([]string{"foo"})                  // S1
-	createStream([]string{"bar"})                  // S2
-	createStream([]string{"baz"})                  // S3
-	createStream([]string{"foo.*", "bar.*"})       // S4
-	createStream([]string{"foo-1.22", "bar-1.33"}) // S5
-
-	expectStreams := func(filter string, streams []string) {
-		t.Helper()
-		req, _ := json.Marshal(&server.JSApiStreamNamesRequest{Subject: filter})
-		r, _ := nc.Request(server.JSApiStreams, req, time.Second)
-		var resp server.JSApiStreamNamesResponse
-		if err := json.Unmarshal(r.Data, &resp); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		if len(resp.Streams) != len(streams) {
-			t.Fatalf("Expected %d results, got %d", len(streams), len(resp.Streams))
-		}
-	}
-
-	expectStreams("foo", []string{"S1"})
-	expectStreams("bar", []string{"S2"})
-	expectStreams("baz", []string{"S3"})
-	expectStreams("*", []string{"S1", "S2", "S3"})
-	expectStreams(">", []string{"S1", "S2", "S3", "S4", "S5"})
-	expectStreams("*.*", []string{"S4", "S5"})
-	expectStreams("*.22", []string{"S4", "S5"})
-}
-
 func TestJetStreamAPIStreamListPaging(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer s.Shutdown()
@@ -7574,7 +7623,7 @@ func TestJetStreamAPIStreamListPaging(t *testing.T) {
 		t.Helper()
 		var req []byte
 		if offset > 0 {
-			req, _ = json.Marshal(&server.ApiPagedRequest{Offset: offset})
+			req, _ = json.Marshal(&server.JSApiStreamNamesRequest{ApiPagedRequest: server.ApiPagedRequest{Offset: offset}})
 		}
 		resp, err := nc.Request(server.JSApiStreams, req, time.Second)
 		if err != nil {
@@ -9203,53 +9252,6 @@ func TestJetStreamAckExplicitMsgRemoval(t *testing.T) {
 	}
 }
 
-// This test is in support fo clients that want to match on subject, they
-// can set the filter subject always and if the stream only has one subject
-// and they match the filter is cleared automatically. This eliminates us
-// needing to know if a subject is a subset of a stream when looking it up.
-func TestJetStreamConsumerFilterSubject(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer s.Shutdown()
-
-	sc := &server.StreamConfig{Name: "MY_STREAM", Subjects: []string{"foo"}}
-	mset, err := s.GlobalAccount().AddStream(sc)
-	if err != nil {
-		t.Fatalf("Unexpected error adding stream: %v", err)
-	}
-	defer mset.Delete()
-
-	cfg := &server.ConsumerConfig{
-		Durable:        "d",
-		DeliverSubject: "A",
-		AckPolicy:      server.AckExplicit,
-		FilterSubject:  "foo",
-	}
-
-	o, err := mset.AddConsumer(cfg)
-	if err != nil {
-		t.Fatalf("Unexpected error adding consumer: %v", err)
-	}
-	defer o.Delete()
-
-	if o.Info().Config.FilterSubject != "" {
-		t.Fatalf("Expected the filter to be cleared")
-	}
-
-	// Now use the original cfg with updated delivery subject and make sure that works ok.
-	cfg = &server.ConsumerConfig{
-		Durable:        "d",
-		DeliverSubject: "B",
-		AckPolicy:      server.AckExplicit,
-		FilterSubject:  "foo",
-	}
-
-	o, err = mset.AddConsumer(cfg)
-	if err != nil {
-		t.Fatalf("Unexpected error adding consumer: %v", err)
-	}
-	defer o.Delete()
-}
-
 func TestJetStreamStoredMsgsDontDisappearAfterCacheExpiration(t *testing.T) {
 	sc := &server.StreamConfig{
 		Name:      "MY_STREAM",