Repository: nats-io/nats-server
Pull request #2322 (2021-06-28 17:58:44)

Added ability to match based on last expected sequence per subject.
Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index deb17280..c3458392 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -3857,7 +3857,7 @@ func (mb *msgBlock) removeSeqPerSubject(subj string, seq uint64) {
 		return
 	}
 	// TODO(dlc) - Might want to optimize this.
-	for tseq := seq + 1; tseq < ss.Last; tseq++ {
+	for tseq := seq + 1; tseq <= ss.Last; tseq++ {
 		if sm, _ := mb.cacheLookupWithLock(tseq); sm != nil {
 			if sm.subj == subj {
 				ss.First = tseq
diff --git a/server/stream.go b/server/stream.go
index e1503956..b5eaf339 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -211,13 +211,14 @@ type sourceInfo struct {
 
 // Headers for published messages.
 const (
-	JSMsgId             = "Nats-Msg-Id"
-	JSExpectedStream    = "Nats-Expected-Stream"
-	JSExpectedLastSeq   = "Nats-Expected-Last-Sequence"
-	JSExpectedLastMsgId = "Nats-Expected-Last-Msg-Id"
-	JSStreamSource      = "Nats-Stream-Source"
-	JSLastConsumerSeq   = "Nats-Last-Consumer"
-	JSLastStreamSeq     = "Nats-Last-Stream"
+	JSMsgId               = "Nats-Msg-Id"
+	JSExpectedStream      = "Nats-Expected-Stream"
+	JSExpectedLastSeq     = "Nats-Expected-Last-Sequence"
+	JSExpectedLastSubjSeq = "Nats-Expected-Last-Subject-Sequence"
+	JSExpectedLastMsgId   = "Nats-Expected-Last-Msg-Id"
+	JSStreamSource        = "Nats-Stream-Source"
+	JSLastConsumerSeq     = "Nats-Last-Consumer"
+	JSLastStreamSeq       = "Nats-Last-Stream"
 )
 
 // Dedupe entry
@@ -2441,6 +2442,15 @@ func getExpectedLastSeq(hdr []byte) uint64 {
 	return uint64(parseInt64(bseq))
 }
 
+// Fast lookup of expected stream sequence per subject.
+func getExpectedLastSeqPerSubject(hdr []byte) uint64 {
+	bseq := getHeader(JSExpectedLastSubjSeq, hdr)
+	if len(bseq) == 0 {
+		return 0
+	}
+	return uint64(parseInt64(bseq))
+}
+
 // Lock should be held.
 func (mset *stream) isClustered() bool {
 	return mset.node != nil
@@ -2650,6 +2660,22 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 			}
 			return fmt.Errorf("last msgid mismatch: %q vs %q", lmsgId, last)
 		}
+		// Expected last sequence per subject.
+		if seq := getExpectedLastSeqPerSubject(hdr); seq > 0 {
+			// TODO(dlc) - We could make a new store func that does this all in one.
+			_, lseq, _, _, _, err := mset.store.LoadLastMsg(subject)
+			if err != nil || lseq != seq {
+				mset.clfs++
+				mset.mu.Unlock()
+				if canRespond {
+					resp.PubAck = &PubAck{Stream: name}
+					resp.Error = ApiErrors[JSStreamWrongLastSequenceErrF].NewT("{seq}", lseq)
+					b, _ := json.Marshal(resp)
+					outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0, nil})
+				}
+				return fmt.Errorf("last sequence by subject mismatch: %d vs %d", seq, lseq)
+			}
+		}
 	}
 
 	// Response Ack.


Test patch:
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index f9c53439..e12edebb 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -10971,6 +10971,71 @@ func TestJetStreamGetLastMsgBySubject(t *testing.T) {
 	}
 }
 
+func TestJetStreamLastSequenceBySubject(t *testing.T) {
+	for _, st := range []StorageType{FileStorage, MemoryStorage} {
+		t.Run(st.String(), func(t *testing.T) {
+			c := createJetStreamClusterExplicit(t, "JSC", 3)
+			defer c.shutdown()
+
+			nc, js := jsClientConnect(t, c.randomServer())
+			defer nc.Close()
+
+			cfg := StreamConfig{
+				Name:       "KV",
+				Subjects:   []string{"kv.>"},
+				Storage:    st,
+				Replicas:   3,
+				MaxMsgsPer: 1,
+			}
+
+			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)
+			}
+
+			js.PublishAsync("kv.foo", []byte("1"))
+			js.PublishAsync("kv.bar", []byte("2"))
+			js.PublishAsync("kv.baz", []byte("3"))
+
+			select {
+			case <-js.PublishAsyncComplete():
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive completion signal")
+			}
+
+			// Now make sure we get an error if the last sequence is not correct per subject.
+			pubAndCheck := func(subj, seq string, ok bool) {
+				t.Helper()
+				m := nats.NewMsg(subj)
+				m.Data = []byte("HELLO")
+				m.Header.Set(JSExpectedLastSubjSeq, seq)
+				_, err := js.PublishMsg(m)
+				if ok && err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				if !ok && err == nil {
+					t.Fatalf("Expected to get an error and got none")
+				}
+			}
+
+			pubAndCheck("kv.foo", "1", true)  // So last is now 4.
+			pubAndCheck("kv.foo", "1", false) // This should fail.
+			pubAndCheck("kv.bar", "2", true)
+			pubAndCheck("kv.bar", "5", true)
+			pubAndCheck("kv.xxx", "5", false)
+		})
+	}
+}
+
 // https://github.com/nats-io/nats-server/issues/2314
 func TestJetStreamMaxMsgsPerAndDiscardNew(t *testing.T) {
 	for _, st := range []StorageType{FileStorage, MemoryStorage} {