Repository: nats-io/nats-server
Pull request #1721 (2020-11-17 23:07:37)

Fix for not properly recovering first sequence number on recovery.
When all messages where deleted we could forget the sequence number that we should be starting back off with on restart/receovery.

This also fixes a few minor other bugs.

/cc @nats-io/core



Patch:
diff --git a/conf/fuzz.go b/conf/fuzz.go
deleted file mode 100644
index 9dce7f7e..00000000
--- a/conf/fuzz.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright 2020 The NATS Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// +build gofuzz
-
-package conf
-
-func Fuzz(data []byte) int {
-	_, err := Parse(string(data))
-	if err != nil {
-		return 0
-	}
-	return 1
-}
diff --git a/server/consumer.go b/server/consumer.go
index e22f7864..afc522ce 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -751,9 +751,9 @@ 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, dcount, _, _ := o.ReplyInfo(subject)
 
-	skipAckReply := sseq == 0
+	var skipAckReply bool
 
 	switch {
 	case len(msg) == 0, bytes.Equal(msg, AckAck), bytes.Equal(msg, AckOK):
@@ -1830,73 +1830,23 @@ func (o *Consumer) checkPending() {
 
 // SeqFromReply will extract a sequence number from a reply subject.
 func (o *Consumer) SeqFromReply(reply string) uint64 {
-	_, dseq, _ := ackReplyInfo(reply)
-	return dseq
+	_, seq, _, _, _ := o.ReplyInfo(reply)
+	return seq
 }
 
 // StreamSeqFromReply will extract the stream sequence from the reply subject.
 func (o *Consumer) StreamSeqFromReply(reply string) uint64 {
-	sseq, _, _ := ackReplyInfo(reply)
-	return sseq
+	seq, _, _, _, _ := o.ReplyInfo(reply)
+	return seq
 }
 
-// Quick parser for positive numbers in ack reply encoding.
-func parseAckReplyNum(d string) (n int64) {
-	if len(d) == 0 {
-		return -1
-	}
-	for _, dec := range d {
-		if dec < asciiZero || dec > asciiNine {
-			return -1
-		}
-		n = n*10 + (int64(dec) - asciiZero)
-	}
-	return n
-}
-
-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) {
-	tsa := [expectedNumReplyTokens]string{}
-	start, tokens := 0, tsa[:0]
-	for i := 0; i < len(subject); i++ {
-		if subject[i] == btsep {
-			tokens = append(tokens, subject[start:i])
-			start = i + 1
-		}
-	}
-	tokens = append(tokens, subject[start:])
-	if len(tokens) != expectedNumReplyTokens || tokens[0] != "$JS" || tokens[1] != "ACK" {
+func (o *Consumer) ReplyInfo(reply string) (sseq, dseq, dcount uint64, ts int64, pending uint64) {
+	n, err := fmt.Sscanf(reply, o.ackReplyT, &dcount, &sseq, &dseq, &ts, &pending)
+	if err != nil || n != 5 {
 		return 0, 0, 0, 0, 0
 	}
-	// TODO(dlc) - Should we error if we do not match consumer name?
-	// stream is tokens[2], consumer is 3.
-	dcount = 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
-}
-
-func ackReplyInfo(subject string) (sseq, dseq, dcount uint64) {
-	tsa := [expectedNumReplyTokens]string{}
-	start, tokens := 0, tsa[:0]
-	for i := 0; i < len(subject); i++ {
-		if subject[i] == btsep {
-			tokens = append(tokens, subject[start:i])
-			start = i + 1
-		}
-	}
-	tokens = append(tokens, subject[start:])
-	if len(tokens) != expectedNumReplyTokens || tokens[0] != "$JS" || tokens[1] != "ACK" {
-		return 0, 0, 0
-	}
-	dcount = uint64(parseAckReplyNum(tokens[4]))
-	sseq, dseq = uint64(parseAckReplyNum(tokens[5])), uint64(parseAckReplyNum(tokens[6]))
-
-	return sseq, dseq, dcount
+	return
 }
 
 // NextSeq returns the next delivered sequence number for this observable.
diff --git a/server/filestore.go b/server/filestore.go
index f5e91100..fd1520fa 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -606,6 +606,10 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	ts := time.Now().UnixNano()
 	mb.llts, mb.lrts, mb.lwts = ts, ts, ts
 
+	// Remember our last sequence number.
+	mb.first.seq = fs.state.LastSeq + 1
+	mb.last.seq = fs.state.LastSeq
+
 	// We know we will need this so go ahead and spin up.
 	mb.spinUpFlushLoop()
 
@@ -1148,6 +1152,7 @@ func (mb *msgBlock) selectNextFirst() {
 	}
 	// Set new first sequence.
 	mb.first.seq = seq
+
 	// Check if we are empty..
 	if mb.isEmpty() {
 		mb.first.ts = 0
@@ -1381,8 +1386,8 @@ func (fs *fileStore) checkAndFlushAllBlocks() {
 	for _, mb := range fs.blks {
 		if mb.pendingWriteSize() > 0 {
 			mb.flushPendingMsgsAndWait()
-			mb.writeIndexInfo()
 		}
+		mb.writeIndexInfo()
 	}
 }
 
@@ -1562,7 +1567,7 @@ func (mb *msgBlock) numBytes() uint64 {
 // Update accounting on a write msg.
 // Lock should be held.
 func (mb *msgBlock) updateAccounting(seq uint64, ts int64, rl uint64) {
-	if mb.first.seq == 0 {
+	if mb.first.seq == 0 || mb.first.ts == 0 {
 		mb.first.seq = seq
 		mb.first.ts = ts
 	}
@@ -2416,7 +2421,7 @@ func (mb *msgBlock) dirtyClose() {
 
 // Should be called with lock held.
 func (mb *msgBlock) dirtyCloseWithRemove(remove bool) {
-	if mb == nil || mb.qch == nil {
+	if mb == nil {
 		return
 	}
 	// Close cache
diff --git a/server/fuzz.go b/server/fuzz.go
deleted file mode 100644
index 1f626983..00000000
--- a/server/fuzz.go
+++ /dev/null
@@ -1,47 +0,0 @@
-// Copyright 2020 The NATS Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// +build gofuzz
-
-package server
-
-var defaultServerOptions = Options{
-	Host:                  "127.0.0.1",
-	Trace:                 true,
-	Debug:                 true,
-	DisableShortFirstPing: true,
-	NoLog:                 true,
-	NoSigs:                true,
-}
-
-func dummyClient() *client {
-	return &client{srv: New(&defaultServerOptions), msubs: -1, mpay: -1, mcl: MAX_CONTROL_LINE_SIZE}
-}
-
-func FuzzClient(data []byte) int {
-	if len(data) < 100 {
-		return -1
-	}
-	c := dummyClient()
-
-	err := c.parse(data[:50])
-	if err != nil {
-		return 0
-	}
-
-	err = c.parse(data[50:])
-	if err != nil {
-		return 0
-	}
-	return 1
-}
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index ba12c443..bdd61221 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -266,11 +266,6 @@ 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
@@ -985,27 +980,14 @@ func (s *Server) jsStreamLookupRequest(sub *subscription, c *client, subject, re
 		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"}
+		resp.Error = &ApiError{Code: 400, Description: "subject argument is not a valid subject"}
 		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
diff --git a/server/monitor.go b/server/monitor.go
index a1e3858b..b89590a8 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1065,7 +1065,7 @@ type GatewayOptsVarz struct {
 	Advertise      string                  `json:"advertise,omitempty"`
 	ConnectRetries int                     `json:"connect_retries,omitempty"`
 	Gateways       []RemoteGatewayOptsVarz `json:"gateways,omitempty"`
-	RejectUnknown  bool                    `json:"reject_unknown,omitempty"` // config got renamed to reject_unknown_cluster
+	RejectUnknown  bool                    `json:"reject_unknown,omitempty"`
 }
 
 // RemoteGatewayOptsVarz contains monitoring remote gateway information
diff --git a/server/opts.go b/server/opts.go
index 817f198d..bd67267b 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -92,7 +92,7 @@ type GatewayOpts struct {
 	Advertise      string               `json:"advertise,omitempty"`
 	ConnectRetries int                  `json:"connect_retries,omitempty"`
 	Gateways       []*RemoteGatewayOpts `json:"gateways,omitempty"`
-	RejectUnknown  bool                 `json:"reject_unknown,omitempty"` // config got renamed to reject_unknown_cluster
+	RejectUnknown  bool                 `json:"reject_unknown,omitempty"`
 
 	// Not exported, for tests.
 	resolver         netResolver
@@ -1285,7 +1285,7 @@ func parseGateway(v interface{}, o *Options, errors *[]error, warnings *[]error)
 				return err
 			}
 			o.Gateway.Gateways = gateways
-		case "reject_unknown", "reject_unknown_cluster":
+		case "reject_unknown":
 			o.Gateway.RejectUnknown = mv.(bool)
 		default:
 			if !tk.IsUsedVariable() {


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index ff5570b5..e9bcb935 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -224,6 +224,41 @@ func TestFileStoreBasicWriteMsgsAndRestore(t *testing.T) {
 	if state.Bytes != expectedSize*2 {
 		t.Fatalf("Expected %d bytes, got %d", expectedSize*2, state.Bytes)
 	}
+
+	fs.Purge()
+	fs.Stop()
+
+	fs, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	state = fs.State()
+	if state.Msgs != 0 {
+		t.Fatalf("Expected %d msgs, got %d", 0, state.Msgs)
+	}
+	if state.Bytes != 0 {
+		t.Fatalf("Expected %d bytes, got %d", 0, state.Bytes)
+	}
+
+	seq, _, err := fs.StoreMsg(subj, nil, []byte("Hello"))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	fs.RemoveMsg(seq)
+
+	fs.Stop()
+	fs, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	state = fs.State()
+	if state.FirstSeq != seq+1 {
+		t.Fatalf("Expected first seq to be %d, got %d", seq+1, state.FirstSeq)
+	}
 }
 
 func TestFileStoreSelectNextFirst(t *testing.T) {
@@ -803,7 +838,7 @@ func TestFileStoreRemoveOutOfOrderRecovery(t *testing.T) {
 
 	state2 := fs.State()
 	if state != state2 {
-		t.Fatalf("Expected receovered states to be the same, got %+v vs %+v\n", state, state2)
+		t.Fatalf("Expected recovered states to be the same, got %+v vs %+v\n", state, state2)
 	}
 
 	if _, _, _, _, err := fs.LoadMsg(1); err != nil {
diff --git a/server/opts_test.go b/server/opts_test.go
index 0153b580..38aa562b 100644
--- a/server/opts_test.go
+++ b/server/opts_test.go
@@ -1859,7 +1859,6 @@ func TestParsingGateways(t *testing.T) {
 		listen: "127.0.0.1:4444"
 		host: "127.0.0.1"
 		port: 4444
-		reject_unknown_cluster: true
 		authorization {
 			user: "ivan"
 			password: "pwd"
@@ -1904,7 +1903,6 @@ func TestParsingGateways(t *testing.T) {
 		Advertise:      "me:1",
 		ConnectRetries: 10,
 		TLSTimeout:     3.0,
-		RejectUnknown:  true,
 	}
 	u1, _ := url.Parse("nats://user1:pwd1@host2:5222")
 	u2, _ := url.Parse("nats://user1:pwd1@host3:6222")
@@ -1971,7 +1969,7 @@ func TestParsingGatewaysErrors(t *testing.T) {
 			`gateway {
 				name: "A"
 				port: -1
-				reject_unknown_cluster: true
+				reject_unknown: true
 				unknown_field: 1
 			}`,
 			"unknown field",
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index f974558f..90d78278 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -425,9 +425,6 @@ func TestJetStreamLookupStreamBySubject(t *testing.T) {
 	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) {
@@ -467,7 +464,6 @@ func TestJetStreamLookupStreamBySubject(t *testing.T) {
 	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)
@@ -487,24 +483,12 @@ func TestJetStreamLookupStreamBySubject(t *testing.T) {
 		}
 	}
 
-	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)
+	checkAPILookup("foo", "1", false)
+	checkAPILookup("boo", "2", true)
+	checkAPILookup("foo.*", "3", true)
+	checkAPILookup("foo.1", "3", true)
+	checkAPILookup("baz.*.*.>", "4", false)
+	checkAPILookup("baz.2.*.>", "4", true)
 }
 
 func TestJetStreamConsumerWithStartTime(t *testing.T) {
@@ -9696,3 +9680,107 @@ func TestJetStreamPullConsumerMaxAckPendingRedeliveries(t *testing.T) {
 		})
 	}
 }
+
+func TestJetStreamDeliveryAfterServerRestart(t *testing.T) {
+	opts := DefaultTestOptions
+	opts.Port = -1
+	opts.JetStream = true
+	s := RunServer(&opts)
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+
+	mset, err := s.GlobalAccount().AddStream(&server.StreamConfig{
+		Name:      "MY_STREAM",
+		Storage:   server.FileStorage,
+		Subjects:  []string{"foo.>"},
+		Retention: server.InterestPolicy,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+	defer mset.Delete()
+
+	nc := clientConnectToServer(t, s)
+	defer nc.Close()
+
+	inbox := nats.NewInbox()
+	o, err := mset.AddConsumer(&server.ConsumerConfig{
+		Durable:        "dur",
+		DeliverSubject: inbox,
+		DeliverPolicy:  server.DeliverNew,
+		AckPolicy:      server.AckExplicit,
+	})
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+	defer o.Delete()
+
+	sub, err := nc.SubscribeSync(inbox)
+	if err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+	nc.Flush()
+
+	// Send 1 message
+	sendStreamMsg(t, nc, "foo.bar", "msg1")
+
+	// Make sure we receive it and ack it.
+	msg, err := sub.NextMsg(250 * time.Millisecond)
+	if err != nil {
+		t.Fatalf("Did not get message: %v", err)
+	}
+	// Ack it!
+	msg.Respond(nil)
+	nc.Flush()
+
+	// Shutdown client and server
+	nc.Close()
+
+	dir := strings.TrimSuffix(s.JetStreamConfig().StoreDir, server.JetStreamStoreDir)
+	s.Shutdown()
+
+	opts.Port = -1
+	opts.StoreDir = dir
+	s = RunServer(&opts)
+	defer s.Shutdown()
+
+	// Lookup stream.
+	mset, err = s.GlobalAccount().LookupStream("MY_STREAM")
+	if err != nil {
+		t.Fatalf("Error looking up stream: %v", err)
+	}
+
+	// Update consumer's deliver subject with new inbox
+	inbox = nats.NewInbox()
+	o, err = mset.AddConsumer(&server.ConsumerConfig{
+		Durable:        "dur",
+		DeliverSubject: inbox,
+		DeliverPolicy:  server.DeliverNew,
+		AckPolicy:      server.AckExplicit,
+	})
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+	defer o.Delete()
+
+	nc = clientConnectToServer(t, s)
+	defer nc.Close()
+
+	// Send 2nd message
+	sendStreamMsg(t, nc, "foo.bar", "msg2")
+
+	// Start sub on new inbox
+	sub, err = nc.SubscribeSync(inbox)
+	if err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+	nc.Flush()
+
+	// Should receive message 2.
+	if _, err := sub.NextMsg(500 * time.Millisecond); err != nil {
+		t.Fatalf("Did not get message: %v", err)
+	}
+}